自学教程：C++ thread_add_read函数代码示例

void smart_link_event(enum sl_event event, int sock){    switch (event)    {        case SL_EVENT:        {            smart_link_log_event("sl-event event, fd(%d)./n", sock);            thread_add_read(sl_master, smart_link_recv_event, NULL, sock);            break;        }        case SL_MSG:        {            smart_link_log_event("sl-event msg, fd(%d)./n", sock);            thread_add_read(sl_master, smart_link_recv_msg, NULL, sock);            break;        }        case SL_DBUS_READ:        {            smart_link_log_event("sl-event dbus-read, fd(%d)./n", sock);            thread_add_read(sl_master, smart_link_recv_cmd, NULL, sock);            break;        }        case SL_DBUS_WRITE:        {            smart_link_log_event("sl-event dbus-write, fd(%d)./n", sock);            thread_add_write(sl_master, smart_link_send_cmd, NULL, sock);            break;        }        default :        {            smart_link_log_error("sl-event default, fd(%d)./n", sock);            break;        }    }}

/* * Read get result from the remote web server. * Apply trigger check to this result. */inthttp_response_thread(thread_t * thread){	SOCK *sock_obj = THREAD_ARG(thread);	/* Handle read timeout */	if (thread->type == THREAD_READ_TIMEOUT)		return epilog(thread);	/* Allocate & clean the get buffer */	sock_obj->buffer = (char *) MALLOC(MAX_BUFFER_LENGTH);	/* Initalize the hash context */	sock_obj->hash = &hashes[req->hash];	HASH_INIT(sock_obj);	/* Register asynchronous http/ssl read thread */	if (req->ssl)		thread_add_read(thread->master, ssl_read_thread, sock_obj,				thread->u.fd, HTTP_CNX_TIMEOUT);	else		thread_add_read(thread->master, http_read_thread, sock_obj,				thread->u.fd, HTTP_CNX_TIMEOUT);	return 0;}

intospf_if_up (struct ospf_interface *oi){  if (oi == NULL)    return 0;  if (oi->type == OSPF_IFTYPE_LOOPBACK)    OSPF_ISM_EVENT_SCHEDULE (oi, ISM_LoopInd);  else    {      //struct ospf *ospf = ospf_lookup ();      struct ospf *ospf = oi->ospf;      if (ospf != NULL)        ospf_adjust_sndbuflen_oisock (ospf, oi->ifp->mtu,oi);      else        ;        //zlog_warn ("%s: ospf_lookup() returned NULL", __func__);      ospf_if_stream_set (oi);      OSPF_ISM_EVENT_SCHEDULE (oi, ISM_InterfaceUp);    }  if(oi->t_packet_read == NULL)    oi->t_packet_read = thread_add_read (master, ospf_read_oisock, oi, oi->packet_fd);  return 1;}

voidkernel_netlink_init(void){	/* Start with a netlink address lookup */	netlink_address_lookup();	/*	 * Prepare netlink kernel broadcast channel	 * subscribtion. We subscribe to LINK and ADDR	 * netlink broadcast messages.	 */	netlink_socket(&nl_kernel, SOCK_NONBLOCK, RTNLGRP_LINK, RTNLGRP_IPV4_IFADDR, RTNLGRP_IPV6_IFADDR, 0);	if (nl_kernel.fd > 0) {		log_message(LOG_INFO, "Registering Kernel netlink reflector");		nl_kernel.thread = thread_add_read(master, kernel_netlink, &nl_kernel, nl_kernel.fd,						   NETLINK_TIMER);	} else		log_message(LOG_INFO, "Error while registering Kernel netlink reflector channel");	/* Prepare netlink command channel. */	netlink_socket(&nl_cmd, SOCK_NONBLOCK, 0);	if (nl_cmd.fd > 0)		log_message(LOG_INFO, "Registering Kernel netlink command channel");	else		log_message(LOG_INFO, "Error while registering Kernel netlink cmd channel");}

static intbabel_create_routing_process (void){    assert (babel_routing_process == NULL);    /* Allocaste Babel instance. */    babel_routing_process = XCALLOC (MTYPE_BABEL, sizeof (struct babel));    /* Initialize timeouts */    gettime(&babel_now);    expiry_time = babel_now.tv_sec + roughly(30);    source_expiry_time = babel_now.tv_sec + roughly(300);    /* Make socket for Babel protocol. */    protocol_socket = babel_socket(protocol_port);    if (protocol_socket < 0) {        zlog_err("Couldn't create link local socket: %s", safe_strerror(errno));        goto fail;    }    /* Threads. */    babel_routing_process->t_read =    thread_add_read(master, &babel_read_protocol, NULL, protocol_socket);    /* wait a little: zebra will announce interfaces, addresses, routes... */    babel_routing_process->t_update =    thread_add_timer_msec(master, babel_init_routing_process, NULL, 200L);    schedule_auth_housekeeping();    return 0;fail:    XFREE(MTYPE_BABEL, babel_routing_process);    babel_routing_process = NULL;    return -1;}

/* Make routing socket. */static voidrouting_socket (void){  if ( zserv_privs.change (ZPRIVS_RAISE) )    zlog_err ("routing_socket: Can't raise privileges");  routing_sock = socket (AF_ROUTE, SOCK_RAW, 0);  if (routing_sock < 0)     {      if ( zserv_privs.change (ZPRIVS_LOWER) )        zlog_err ("routing_socket: Can't lower privileges");      zlog_warn ("Can't init kernel routing socket");      return;    }  /* XXX: Socket should be NONBLOCK, however as we currently    * discard failed writes, this will lead to inconsistencies.   * For now, socket must be blocking.   */  /*if (fcntl (routing_sock, F_SETFL, O_NONBLOCK) < 0)     zlog_warn ("Can't set O_NONBLOCK to routing socket");*/      if ( zserv_privs.change (ZPRIVS_LOWER) )    zlog_err ("routing_socket: Can't lower privileges");  /* kernel_read needs rewrite. */  thread_add_read (zebrad.master, kernel_read, NULL, routing_sock);}

/* thread reading entries form others babel daemons */static intbabel_read_protocol (struct thread *thread){    int rc;    struct interface *ifp = NULL;    struct sockaddr_in6 sin6;    struct listnode *linklist_node = NULL;    assert(babel_routing_process != NULL);    assert(protocol_socket >= 0);    rc = babel_recv(protocol_socket,                    receive_buffer, receive_buffer_size,                    (struct sockaddr*)&sin6, sizeof(sin6));    if(rc < 0) {        if(errno != EAGAIN && errno != EINTR) {            zlog_err("recv: %s", safe_strerror(errno));        }    } else {        FOR_ALL_INTERFACES(ifp, linklist_node) {            if(!if_up(ifp))                continue;            if(ifp->ifindex == sin6.sin6_scope_id) {                parse_packet((unsigned char*)&sin6.sin6_addr, ifp,                             receive_buffer, rc);                break;            }        }    }    /* re-add thread */    babel_routing_process->t_read =    thread_add_read(master, &babel_read_protocol, NULL, protocol_socket);    return 0;}

voidkernel_netlink_init(void){	unsigned long groups;	/* Start with a netlink address lookup */	netlink_address_lookup();	/*	 * Prepare netlink kernel broadcast channel	 * subscribtion. We subscribe to LINK and ADDR	 * netlink broadcast messages.	 */	groups = RTMGRP_LINK | RTMGRP_IPV4_IFADDR | RTMGRP_IPV6_IFADDR;	netlink_socket(&nl_kernel, groups);	if (nl_kernel.fd > 0) {		log_message(LOG_INFO, "Registering Kernel netlink reflector");		thread_add_read(master, kernel_netlink, NULL, nl_kernel.fd,				NETLINK_TIMER);	} else		log_message(LOG_INFO, "Error while registering Kernel netlink reflector channel");	/* Prepare netlink command channel. */	netlink_socket(&nl_cmd, 0);	if (nl_cmd.fd > 0)		log_message(LOG_INFO, "Registering Kernel netlink command channel");	else		log_message(LOG_INFO, "Error while registering Kernel netlink cmd channel");}

static voidzclient_event (enum event event, struct zclient *zclient){  switch (event)    {    case ZCLIENT_SCHEDULE:      if (! zclient->t_connect)	zclient->t_connect =	  thread_add_event (master, zclient_connect, zclient, 0);      break;    case ZCLIENT_CONNECT:      if (zclient->fail >= 10)	return;      if (zclient_debug)	zlog_debug ("zclient connect schedule interval is %d", 		   zclient->fail < 3 ? 10 : 60);      if (! zclient->t_connect)	zclient->t_connect = 	  thread_add_timer (master, zclient_connect, zclient,			    zclient->fail < 3 ? 10 : 60);      break;    case ZCLIENT_READ:      zclient->t_read = 	thread_add_read (master, zclient_read, zclient, zclient->sock);      break;    }}

static intsmtp_send_thread(thread_t * thread){	smtp_t *smtp = THREAD_ARG(thread);	if (thread->type == THREAD_WRITE_TIMEOUT) {		log_message(LOG_INFO, "Timeout sending data to remote SMTP server %s."				    , FMT_SMTP_HOST());		SMTP_FSM_READ(QUIT, thread, 0);		return 0;	}	SMTP_FSM_SEND(smtp->stage, thread);	/* Handle END command */	if (smtp->stage == END) {		SMTP_FSM_READ(QUIT, thread, 0);		return 0;	}	/* Registering next smtp command processing thread */	if (smtp->stage != ERROR) {		thread_add_read(thread->master, smtp_read_thread, smtp,				thread->u.fd, global_data->smtp_connection_to);	} else {		log_message(LOG_INFO, "Can not send data to remote SMTP server %s."				    , FMT_SMTP_HOST());		SMTP_FSM_READ(QUIT, thread, 0);	}	return 0;}

static void ext_client_bgp_event(enum event event, struct ext_client_bgp * ext_client_bgp, struct bgp_listener * listener){  switch(event)  {     case EXT_CLIENT_BGP_ACCEPT:      listener->t_accept = thread_add_read(master, ext_client_bgp_accept, ext_client_bgp, listener->accept_fd);      break;    case EXT_CLIENT_BGP_READ:      listener->t_read = thread_add_read(master, ext_client_bgp_recv, ext_client_bgp, listener->peer_fd);      break;    default:      break;  }}

/* Kernel route reflection. */static intkernel_read (struct thread *thread){  netlink_parse_info (netlink_information_fetch, &netlink);  thread_add_read (hm->master, kernel_read, NULL, netlink.sock);  return 0;}

static voidzebra_event (enum event event, int sock, struct zserv *client){  switch (event)    {    case ZEBRA_SERV:      thread_add_read (zebrad.master, zebra_accept, client, sock);      break;    case ZEBRA_READ:      client->t_read = 	thread_add_read (zebrad.master, zebra_client_read, client, sock);      break;    case ZEBRA_WRITE:      /**/      break;    }}

/* Filter out messages from self that occur on listener socket,   caused by our actions on the command socket */static void netlink_install_filter (int sock, __u32 pid){  struct sock_filter filter[] = {    /* 0: ldh [4]	          */    BPF_STMT(BPF_LD|BPF_ABS|BPF_H, offsetof(struct nlmsghdr, nlmsg_type)),    /* 1: jeq 0x18 jt 3 jf 6  */    BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K, htons(RTM_NEWROUTE), 1, 0),    /* 2: jeq 0x19 jt 3 jf 6  */    BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K, htons(RTM_DELROUTE), 0, 3),    /* 3: ldw [12]		  */    BPF_STMT(BPF_LD|BPF_ABS|BPF_W, offsetof(struct nlmsghdr, nlmsg_pid)),    /* 4: jeq XX  jt 5 jf 6   */    BPF_JUMP(BPF_JMP|BPF_JEQ|BPF_K, htonl(pid), 0, 1),    /* 5: ret 0    (skip)     */    BPF_STMT(BPF_RET|BPF_K, 0),    /* 6: ret 0xffff (keep)   */    BPF_STMT(BPF_RET|BPF_K, 0xffff),  };  struct sock_fprog prog = {    .len = sizeof(filter) / sizeof(filter[0]),    .filter = filter,  };  if (setsockopt(sock, SOL_SOCKET, SO_ATTACH_FILTER, &prog, sizeof(prog)) < 0)    zlog_warn ("Can't install socket filter: %s/n", safe_strerror(errno));}/* Exported interface function.  This function simply calls   netlink_socket (). */voidkernel_init (void){  unsigned long groups;  groups = RTMGRP_LINK | RTMGRP_IPV4_ROUTE | RTMGRP_IPV4_IFADDR;#ifdef HAVE_IPV6  groups |= RTMGRP_IPV6_ROUTE | RTMGRP_IPV6_IFADDR;#endif /* HAVE_IPV6 */  netlink_socket (&netlink, groups);  netlink_socket (&netlink_cmd, 0);  /* Register kernel socket. */  if (netlink.sock > 0)    {      /* Only want non-blocking on the netlink event socket */      if (fcntl (netlink.sock, F_SETFL, O_NONBLOCK) < 0)	zlog (NULL, LOG_ERR, "Can't set %s socket flags: %s", netlink.name,		safe_strerror (errno));      /* Set receive buffer size if it's set from command line */      if (nl_rcvbufsize)	netlink_recvbuf (&netlink, nl_rcvbufsize);      netlink_install_filter (netlink.sock, netlink_cmd.snl.nl_pid);      thread_add_read (hm->master, kernel_read, NULL, netlink.sock);    }}

intkernel_netlink(thread_t * thread){	if (thread->type != THREAD_READ_TIMEOUT)		netlink_parse_info(netlink_broadcast_filter, &nl_kernel, NULL);	thread_add_read(master, kernel_netlink, NULL, nl_kernel.fd,			NETLINK_TIMER);	return 0;}

mpls_return_enum mpls_socket_readlist_add(mpls_socket_mgr_handle handle,  mpls_socket_handle socket, void *extra, mpls_socket_enum type){  socket->type = type;  socket->extra = extra;  MPLS_ASSERT(socket && (socket->fd > -1));  socket->read = thread_add_read(master,mplsd_read,socket,socket->fd);  MPLS_ASSERT(socket->read);  return MPLS_SUCCESS;}

/* Traditional IPv4 only version.  */intbgp_socket (struct bgp *bgp, unsigned short port){  int sock;  int socklen;  struct sockaddr_in sin;  int ret;  sock = socket (AF_INET, SOCK_STREAM, 0);  if (sock < 0)    {      zlog_err ("socket: %s", strerror (errno));      return sock;    }  sockopt_reuseaddr (sock);  sockopt_reuseport (sock);  memset (&sin, 0, sizeof (struct sockaddr_in));  sin.sin_family = AF_INET;  sin.sin_port = htons (port);  socklen = sizeof (struct sockaddr_in);#ifdef HAVE_SIN_LEN  sin.sin_len = socklen;#endif /* HAVE_SIN_LEN */  ret = bind (sock, (struct sockaddr *) &sin, socklen);  if (ret < 0)    {      zlog_err ("bind: %s", strerror (errno));      close (sock);      return ret;    }  ret = listen (sock, 3);  if (ret < 0)     {      zlog_err ("listen: %s", strerror (errno));      close (sock);      return ret;    }#ifdef HAVE_TCP_SIGNATURE#ifdef HAVE_LINUX_TCP_SIGNATURE  bm->sock = sock;#endif /* HAVE_LINUX_TCP_SIGNATURE */#ifdef HAVE_OPENBSD_TCP_SIGNATURE  bgp_tcpsig_set (sock, 0);  bm->sock = -1;#endif /* HAVE_OPENBSD_TCP_SIGNATURE */#endif /* HAVE_TCP_SIGNATURE */  thread_add_read (bm->master, bgp_accept, bgp, sock);  return sock;}

static intkernel_netlink(thread_t * thread){	nl_handle_t *nl = THREAD_ARG(thread);	if (thread->type != THREAD_READ_TIMEOUT)		netlink_parse_info(netlink_broadcast_filter, nl, NULL);	nl->thread = thread_add_read(master, kernel_netlink, nl, nl->fd,				      NETLINK_TIMER);	return 0;}

static intbgp_listener (int sock, struct sockaddr *sa, socklen_t salen){  struct bgp_listener *listener;  int ret, en;  sockopt_reuseaddr (sock);  sockopt_reuseport (sock);#ifdef IPTOS_PREC_INTERNETCONTROL  if (sa->sa_family == AF_INET)    setsockopt_ipv4_tos (sock, IPTOS_PREC_INTERNETCONTROL);#endif#ifdef IPV6_V6ONLY  /* Want only IPV6 on ipv6 socket (not mapped addresses) */  if (sa->sa_family == AF_INET6) {    int on = 1;    setsockopt (sock, IPPROTO_IPV6, IPV6_V6ONLY,		(void *) &on, sizeof (on));  }#endif  if (bgpd_privs.change (ZPRIVS_RAISE) )    zlog_err ("bgp_socket: could not raise privs");  ret = bind (sock, sa, salen);  en = errno;  if (bgpd_privs.change (ZPRIVS_LOWER) )    zlog_err ("bgp_bind_address: could not lower privs");  if (ret < 0)    {      zlog_err ("bind: %s", safe_strerror (en));      return ret;    }  ret = listen (sock, 3);  if (ret < 0)    {      zlog_err ("listen: %s", safe_strerror (errno));      return ret;    }  listener = XMALLOC (MTYPE_BGP_LISTENER, sizeof(*listener));  listener->fd = sock;  memcpy(&listener->su, sa, salen);  listener->thread = thread_add_read (master, bgp_accept, listener, sock);  listnode_add (bm->listen_sockets, listener);  return 0;}

static intconnection_success(thread_t * thread){	smtp_t *smtp = THREAD_ARG(thread);	log_message(LOG_INFO, "Remote SMTP server [%s]:%d connected."			    , inet_sockaddrtos(&global_data->smtp_server), SMTP_PORT);	smtp->stage = connect_success;	thread_add_read(thread->master, smtp_read_thread, smtp,			smtp->fd, global_data->smtp_connection_to);	return 0;}

static intconnection_success(thread_t * thread){	smtp_t *smtp = THREAD_ARG(thread);	log_message(LOG_INFO, "Remote SMTP server %s connected."			    , FMT_SMTP_HOST());	smtp->stage = connect_success;	thread_add_read(thread->master, smtp_read_thread, smtp,			smtp->fd, global_data->smtp_connection_to);	return 0;}

intshim_sisis_accept(struct thread * thread){  int accept_sock;  int sisis_sock;  struct sisis_listener *listener;  union sockunion su;  char buf[SU_ADDRSTRLEN];  accept_sock = THREAD_FD (thread);  if (accept_sock < 0)  {    zlog_err ("accept_sock is negative value %d", accept_sock);    return -1;  }  thread_add_read (master, shim_sisis_accept, NULL, accept_sock);  sisis_sock = sockunion_accept(accept_sock, &su);  if (sisis_sock < 0)  {    zlog_err ("[Error] SISIS socket accept failed (%s)", safe_strerror (errno));    return -1;  }  zlog_notice ("SISIS connection from host %s", inet_sutop (&su, buf));   listener = XMALLOC (MTYPE_SHIM_SISIS_LISTENER, sizeof(*listener));  listener->fd = accept_sock;  listener->ibuf = stream_new (SV_HEADER_SIZE + 1500);//  memcpy(&listener->su, sa, salen);  listener->sisis_fd = sisis_sock;  listener->dif = stream_fifo_new();  listener->chksum_stream = stream_new(4 * 20); // need to figure out good size for buffering  listener->read_thread = thread_add_read (master, shim_sisis_read, listener, sisis_sock);  listnode_add (sm->listen_sockets, listener);  return 0;}

static int bgp_listener(int sock, struct sockaddr *sa, socklen_t salen,			struct bgp *bgp){	struct bgp_listener *listener;	int ret, en;	sockopt_reuseaddr(sock);	sockopt_reuseport(sock);	frr_elevate_privs(&bgpd_privs) {#ifdef IPTOS_PREC_INTERNETCONTROL		if (sa->sa_family == AF_INET)			setsockopt_ipv4_tos(sock, IPTOS_PREC_INTERNETCONTROL);		else if (sa->sa_family == AF_INET6)			setsockopt_ipv6_tclass(sock,					       IPTOS_PREC_INTERNETCONTROL);#endif		sockopt_v6only(sa->sa_family, sock);		ret = bind(sock, sa, salen);		en = errno;	}	if (ret < 0) {		flog_err_sys(EC_LIB_SOCKET, "bind: %s", safe_strerror(en));		return ret;	}	ret = listen(sock, SOMAXCONN);	if (ret < 0) {		flog_err_sys(EC_LIB_SOCKET, "listen: %s", safe_strerror(errno));		return ret;	}	listener = XCALLOC(MTYPE_BGP_LISTENER, sizeof(*listener));	listener->fd = sock;	/* this socket needs a change of ns. record bgp back pointer */	if (bgp->vrf_id != VRF_DEFAULT && vrf_is_backend_netns())		listener->bgp = bgp;	memcpy(&listener->su, sa, salen);	listener->thread = NULL;	thread_add_read(bm->master, bgp_accept, listener, sock,			&listener->thread);	listnode_add(bm->listen_sockets, listener);	return 0;}

static intbgp_listener (int sock, struct sockaddr *sa, socklen_t salen){  struct bgp_listener *listener;  int ret, en;  sockopt_reuseaddr (sock);  sockopt_reuseport (sock);  if (bgpd_privs.change (ZPRIVS_RAISE))    zlog_err ("%s: could not raise privs", __func__);#ifdef IPTOS_PREC_INTERNETCONTROL  if (sa->sa_family == AF_INET)    setsockopt_ipv4_tos (sock, IPTOS_PREC_INTERNETCONTROL);#  ifdef HAVE_IPV6  else if (sa->sa_family == AF_INET6)    setsockopt_ipv6_tclass (sock, IPTOS_PREC_INTERNETCONTROL);#  endif#endif  sockopt_v6only (sa->sa_family, sock);  ret = bind (sock, sa, salen);  en = errno;  if (bgpd_privs.change (ZPRIVS_LOWER))    zlog_err ("%s: could not lower privs", __func__);  if (ret < 0)    {      zlog_err ("bind: %s", safe_strerror (en));      return ret;    }  ret = listen (sock, 3);  if (ret < 0)    {      zlog_err ("listen: %s", safe_strerror (errno));      return ret;    }  listener = XMALLOC (MTYPE_BGP_LISTENER, sizeof(*listener));  listener->fd = sock;  memcpy(&listener->su, sa, salen);  listener->thread = thread_add_read (master, bgp_accept, listener, sock);  listnode_add (bm->listen_sockets, listener);  return 0;}

int irdp_read_raw(struct thread *r){	struct interface *ifp;	struct zebra_if *zi;	struct irdp_interface *irdp;	char buf[IRDP_RX_BUF];	int ret, ifindex = 0;	int irdp_sock = THREAD_FD(r);	t_irdp_raw =	    thread_add_read(zebrad.master, irdp_read_raw, NULL, irdp_sock);	ret = irdp_recvmsg(irdp_sock, (u_char *) buf, IRDP_RX_BUF, &ifindex);	if (ret < 0)		zlog_warn("IRDP: RX Error length = %d", ret);	ifp = if_lookup_by_index(ifindex);	if (!ifp)		return ret;	zi = ifp->info;	if (!zi)		return ret;	irdp = &zi->irdp;	if (!irdp)		return ret;	if (!(irdp->flags & IF_ACTIVE)) {		if (irdp->flags & IF_DEBUG_MISC)			zlog_debug("IRDP: RX ICMP for disabled interface %s/n",				   ifp->name);		return 0;	}	if (irdp->flags & IF_DEBUG_PACKET) {		int i;		zlog_debug("IRDP: RX (idx %d) ", ifindex);		for (i = 0; i < ret; i++)			zlog_debug("IRDP: RX %x ", buf[i] & 0xFF);	}	parse_irdp_packet(buf, ret, ifp);	return ret;}

/* Asynchronous HTTP stream reader */inthttp_read_thread(thread_t * thread){	SOCK *sock_obj = THREAD_ARG(thread);	int r = 0;	/* Handle read timeout */	if (thread->type == THREAD_READ_TIMEOUT)		return epilog(thread);	/* read the HTTP stream */	r = MAX_BUFFER_LENGTH - sock_obj->size;	if (r <= 0) {		/* defensive check, should not occur */		fprintf(stderr, "HTTP socket buffer overflow (not consumed)/n");		r = MAX_BUFFER_LENGTH;	}	memset(sock_obj->buffer + sock_obj->size, 0, r);	r = read(thread->u.fd, sock_obj->buffer + sock_obj->size, r);	DBG(" [l:%d,fd:%d]/n", r, sock_obj->fd);	if (r == -1 || r == 0) {	/* -1:error , 0:EOF */		if (r == -1) {			/* We have encourred a real read error */			DBG("Read error with server [%s]:%d: %s/n",			    req->ipaddress, ntohs(req->addr_port),			    strerror(errno));			return epilog(thread);		}		/* All the HTTP stream has been parsed */		finalize(thread);	} else {		/* Handle the response stream */		http_process_stream(sock_obj, r);		/*		 * Register next http stream reader.		 * Register itself to not perturbe global I/O multiplexer.		 */		thread_add_read(thread->master, http_read_thread, sock_obj,				thread->u.fd, HTTP_CNX_TIMEOUT);	}	return 0;}

/* Make routing socket. */voidrouting_socket (){  routing_sock = socket (AF_ROUTE, SOCK_RAW, 0);  if (routing_sock < 0)     {      zlog_warn ("Can't init kernel routing socket");      return;    }  if (fcntl (routing_sock, F_SETFL, O_NONBLOCK) < 0)     zlog_warn ("Can't set O_NONBLOCK to routing socket");  /* kernel_read needs rewrite. */  thread_add_read (master, kernel_read, NULL, routing_sock);}

int start_comm_salt(thread_t * thread){	struct sockaddr_in srcaddr, dstaddr;      int                sock_fd;    int                flags;    int ret = -1;	sock_fd = socket(AF_INET, SOCK_DGRAM, 0);    if(sock_fd == -1)    {         log_message(LOG_ERR, "not able to create socket: %s", ipvs_strerror(errno));        goto thread_add;    }	srcaddr.sin_family      =  AF_INET;    srcaddr.sin_port        =  htons(35938);    srcaddr.sin_addr.s_addr =  INADDR_LOOPBACK;    if(bind(sock_fd,(struct sockaddr *)&srcaddr, sizeof(srcaddr)) == -1)    {        close(sock_fd);        log_message(LOG_ERR, "binding salt error: %s", ipvs_strerror(errno));        goto thread_add;    }	//set nonblock	flags = fcntl(sock_fd, F_GETFL, 0);	fcntl(sock_fd, F_SETFL, flags | O_NONBLOCK);	dstaddr.sin_family		=  AF_INET;	dstaddr.sin_port		=  htons(35937);	dstaddr.sin_addr.s_addr =  INADDR_LOOPBACK;	ret = connect(sock_fd, (struct sockaddr *)&dstaddr, sizeof(dstaddr));thread_add:    if(ret < 0)		thread_add_timer(master, start_comm_salt, NULL, BOOTSTRAP_DELAY);	else	{	    salt_sock_fd = sock_fd;		thread_add_read(master, lbd_rcvmsg_salt, 0, sock_fd, BOOTSTRAP_DELAY);	}		return 0;