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

/* * this function disables health-check servers so that the process will quickly be ignored * by load balancers. Note that if a proxy was already in the PAUSED state, then its grace * time will not be used since it would already not listen anymore to the socket. */void soft_stop(void){	struct proxy *p;	struct peers *prs;	stopping = 1;	p = proxy;	tv_update_date(0,1); /* else, the old time before select will be used */	while (p) {		if (p->state != PR_STSTOPPED) {			Warning("Stopping %s %s in %d ms./n", proxy_cap_str(p->cap), p->id, p->grace);			send_log(p, LOG_WARNING, "Stopping %s %s in %d ms./n", proxy_cap_str(p->cap), p->id, p->grace);			p->stop_time = tick_add(now_ms, p->grace);		}		if (p->table.size && p->table.sync_task)			 task_wakeup(p->table.sync_task, TASK_WOKEN_MSG);		/* wake every proxy task up so that they can handle the stopping */		task_wakeup(p->task, TASK_WOKEN_MSG);		p = p->next;	}	prs = peers;	while (prs) {		stop_proxy((struct proxy *)prs->peers_fe);		prs = prs->next;	}	/* signal zero is used to broadcast the "stopping" event */	signal_handler(0);}

/* Register a function to handle a stream_interface as a standalone task. The * new task itself is returned and is assigned as si->owner. The stream_interface * pointer will be pointed to by the task's context. The handler can be detached * by using stream_int_unregister_handler(). */struct task *stream_int_register_handler_task(struct stream_interface *si,					      struct task *(*fct)(struct task *)){	struct task *t;	DPRINTF(stderr, "registering handler %p for si %p (was %p)/n", fct, si, si->owner);	si->update  = stream_int_update;	si->shutr   = stream_int_shutr;	si->shutw   = stream_int_shutw;	si->chk_rcv = stream_int_chk_rcv;	si->chk_snd = stream_int_chk_snd;	si->connect = NULL;	si->iohandler = NULL; /* not used when running as an external task */	si->flags |= SI_FL_WAIT_DATA;	t = task_new();	si->owner = t;	if (!t)		return t;	t->process = fct;	t->context = si;	task_wakeup(si->owner, TASK_WOKEN_INIT);	return t;}

/* default shutw function for scheduled tasks */void stream_int_shutw(struct stream_interface *si){	DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x/n",		__FUNCTION__,		si, si->state, si->ib->flags, si->ob->flags);	si->ob->flags &= ~BF_SHUTW_NOW;	if (si->ob->flags & BF_SHUTW)		return;	si->ob->flags |= BF_SHUTW;	si->ob->wex = TICK_ETERNITY;	si->flags &= ~SI_FL_WAIT_DATA;	switch (si->state) {	case SI_ST_EST:		if (!(si->ib->flags & (BF_SHUTR|BF_DONT_READ)))			break;		/* fall through */	case SI_ST_CON:	case SI_ST_CER:		si->state = SI_ST_DIS;		/* fall through */	default:		si->flags &= ~SI_FL_WAIT_ROOM;		si->ib->flags |= BF_SHUTR;		si->ib->rex = TICK_ETERNITY;		si->exp = TICK_ETERNITY;	}	/* note that if the task exist, it must unregister itself once it runs */	if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)		task_wakeup(si->owner, TASK_WOKEN_IO);}

/* * Use this function to force a close of a peer session */static void peer_session_forceshutdown(struct stream * stream){	struct appctx *appctx = NULL;	struct peer_session *ps;	int i;	for (i = 0; i <= 1; i++) {		appctx = objt_appctx(stream->si[i].end);		if (!appctx)			continue;		if (appctx->applet != &peer_applet)			continue;		break;	}	if (!appctx)		return;	ps = (struct peer_session *)appctx->ctx.peers.ptr;	/* we're killing a connection, we must apply a random delay before	 * retrying otherwise the other end will do the same and we can loop	 * for a while.	 */	if (ps)		ps->reconnect = tick_add(now_ms, MS_TO_TICKS(50 + random() % 2000));	/* call release to reinit resync states if needed */	peer_session_release(appctx);	appctx->st0 = PEER_SESS_ST_END;	appctx->ctx.peers.ptr = NULL;	task_wakeup(stream->task, TASK_WOKEN_MSG);}

/* * Callback to release a session with a peer */static void peer_session_release(struct stream_interface *si){	struct task *t = (struct task *)si->owner;	struct session *s = (struct session *)t->context;	struct peer_session *ps = (struct peer_session *)si->conn.data_ctx;	/* si->conn.data_ctx is not a peer session */	if (si->applet.st0 < PEER_SESSION_SENDSUCCESS)		return;	/* peer session identified */	if (ps) {		if (ps->session == s) {			ps->session = NULL;			if (ps->flags & PEER_F_LEARN_ASSIGN) {				/* unassign current peer for learning */				ps->flags &= ~(PEER_F_LEARN_ASSIGN);				ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);				/* reschedule a resync */				ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));			}			/* reset teaching and learning flags to 0 */			ps->flags &= PEER_TEACH_RESET;			ps->flags &= PEER_LEARN_RESET;		}		task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);	}}

/* * Callback to release a session with a peer */static void peer_session_release(struct appctx *appctx){	struct stream_interface *si = appctx->owner;	struct stream *s = si_strm(si);	struct peer_session *ps = (struct peer_session *)appctx->ctx.peers.ptr;	/* appctx->ctx.peers.ptr is not a peer session */	if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)		return;	/* peer session identified */	if (ps) {		if (ps->stream == s) {			ps->stream = NULL;			ps->appctx = NULL;			if (ps->flags & PEER_F_LEARN_ASSIGN) {				/* unassign current peer for learning */				ps->flags &= ~(PEER_F_LEARN_ASSIGN);				ps->table->flags &= ~(SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);				/* reschedule a resync */				ps->table->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));			}			/* reset teaching and learning flags to 0 */			ps->flags &= PEER_TEACH_RESET;			ps->flags &= PEER_LEARN_RESET;		}		task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);	}}

/* * send a message to mail box *   mbox: mail box *   msg: message *   return: RERROR on failed; ROK on success */int mbox_post(mbox_t *mbox, uint32_t *msg){	int ret = RERROR;	uint32_t f;	struct dtask *t;	if(mbox && (mbox->flag & IPC_FLAG_VALID))	{		SYS_FSAVE(f);		if(queue_buffer_put(&mbox->buf, msg))		{			ret = ROK;			t = blockq_select(&(mbox->taskq));			if(t)			{#ifdef INCLUDE_JOURNAL				journal_ipc_post((ipc_t *)mbox, t);#endif // INCLUDE_JOURNAL				t->wakeup_cause = ROK;				task_wakeup(t-systask);			}			else			{#ifdef INCLUDE_JOURNAL				journal_ipc_post((ipc_t *)mbox, t);#endif // INCLUDE_JOURNAL			}		}		SYS_FRESTORE(f);	}	return ret;}

/* default chk_snd function for scheduled tasks */static void stream_int_chk_snd(struct stream_interface *si){	struct channel *oc = si_oc(si);	DPRINTF(stderr, "%s: si=%p, si->state=%d ic->flags=%08x oc->flags=%08x/n",		__FUNCTION__,		si, si->state, si_ic(si)->flags, oc->flags);	if (unlikely(si->state != SI_ST_EST || (oc->flags & CF_SHUTW)))		return;	if (!(si->flags & SI_FL_WAIT_DATA) ||        /* not waiting for data */	    channel_is_empty(oc))           /* called with nothing to send ! */		return;	/* Otherwise there are remaining data to be sent in the buffer,	 * so we tell the handler.	 */	si->flags &= ~SI_FL_WAIT_DATA;	if (!tick_isset(oc->wex))		oc->wex = tick_add_ifset(now_ms, oc->wto);	if (!(si->flags & SI_FL_DONT_WAKE))		task_wakeup(si_task(si), TASK_WOKEN_IO);}

/* * Use this function to force a close of a peer session */static void peer_session_forceshutdown(struct stream * stream){	struct stream_interface *oldsi = NULL;	struct appctx *appctx = NULL;	int i;	for (i = 0; i <= 1; i++) {		appctx = objt_appctx(stream->si[i].end);		if (!appctx)			continue;		if (appctx->applet != &peer_applet)			continue;		oldsi = &stream->si[i];		break;	}	if (!appctx)		return;	/* call release to reinit resync states if needed */	peer_session_release(oldsi);	appctx->st0 = PEER_SESS_ST_END;	appctx->ctx.peers.ptr = NULL;	task_wakeup(stream->task, TASK_WOKEN_MSG);}

/* Check for pending connections at the backend, and assign some of them to * the server coming up. The server's weight is checked before being assigned * connections it may not be able to handle. The total number of transferred * connections is returned. */int pendconn_grab_from_px(struct server *s){	struct pendconn *p;	int maxconn, xferred = 0;	if (!srv_currently_usable(s))		return 0;	/* if this is a backup server and there are active servers or at	 * least another backup server was elected, then this one must	 * not dequeue requests from the proxy.	 */	if ((s->flags & SRV_F_BACKUP) &&	    (s->proxy->srv_act ||	     ((s != s->proxy->lbprm.fbck) && !(s->proxy->options & PR_O_USE_ALL_BK))))		return 0;	HA_SPIN_LOCK(PROXY_LOCK, &s->proxy->lock);	maxconn = srv_dynamic_maxconn(s);	while ((p = pendconn_first(&s->proxy->pendconns))) {		if (s->maxconn && s->served + xferred >= maxconn)			break;		__pendconn_unlink(p);		p->target = s;		task_wakeup(p->strm->task, TASK_WOKEN_RES);		xferred++;	}	HA_SPIN_UNLOCK(PROXY_LOCK, &s->proxy->lock);	return xferred;}

/* default shutr function for scheduled tasks */void stream_int_shutr(struct stream_interface *si){	DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x/n",		__FUNCTION__,		si, si->state, si->ib->flags, si->ob->flags);	si->ib->flags &= ~BF_SHUTR_NOW;	if (si->ib->flags & BF_SHUTR)		return;	si->ib->flags |= BF_SHUTR;	si->ib->rex = TICK_ETERNITY;	si->flags &= ~SI_FL_WAIT_ROOM;	if (si->state != SI_ST_EST && si->state != SI_ST_CON)		return;	if (si->ob->flags & BF_SHUTW) {		si->state = SI_ST_DIS;		si->exp = TICK_ETERNITY;		if (si->release)			si->release(si);	}	/* note that if the task exist, it must unregister itself once it runs */	if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)		task_wakeup(si->owner, TASK_WOKEN_IO);}

/* default chk_snd function for scheduled tasks */void stream_int_chk_snd(struct stream_interface *si){	struct buffer *ob = si->ob;	DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x/n",		__FUNCTION__,		si, si->state, si->ib->flags, si->ob->flags);	if (unlikely(si->state != SI_ST_EST || (si->ob->flags & BF_SHUTW)))		return;	if (!(si->flags & SI_FL_WAIT_DATA) ||        /* not waiting for data */	    (ob->flags & BF_OUT_EMPTY))              /* called with nothing to send ! */		return;	/* Otherwise there are remaining data to be sent in the buffer,	 * so we tell the handler.	 */	si->flags &= ~SI_FL_WAIT_DATA;	if (!tick_isset(ob->wex))		ob->wex = tick_add_ifset(now_ms, ob->wto);	if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)		task_wakeup(si->owner, TASK_WOKEN_IO);}

/* * Function used to register a table for sync on a group of peers * */void peers_register_table(struct peers *peers, struct stktable *table){	struct shared_table *st;	struct peer * curpeer;	struct peer_session *ps;	st = (struct shared_table *)calloc(1,sizeof(struct shared_table));	st->table = table;	st->next = peers->tables;	st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));	peers->tables = st;	for (curpeer = peers->remote; curpeer; curpeer = curpeer->next) {		ps = (struct peer_session *)calloc(1,sizeof(struct peer_session));		ps->table = st;		ps->peer = curpeer;		if (curpeer->local)			st->local_session = ps;		ps->next = st->sessions;		ps->reconnect = now_ms;		st->sessions = ps;		peers->peers_fe->maxconn += 3;	}	peers->peers_fe->listen->maxconn = peers->peers_fe->maxconn;	st->sync_task = task_new();	st->sync_task->process = process_peer_sync;	st->sync_task->expire = TICK_ETERNITY;	st->sync_task->context = (void *)st;	table->sync_task =st->sync_task;	signal_register_task(0, table->sync_task, 0);	task_wakeup(st->sync_task, TASK_WOKEN_INIT);}

/* * This function performs a shutdown-read on a stream interface attached to an * applet in a connected or init state (it does nothing for other states). It * either shuts the read side or marks itself as closed. The buffer flags are * updated to reflect the new state. If the stream interface has SI_FL_NOHALF, * we also forward the close to the write side. The owner task is woken up if * it exists. */static void stream_int_shutr(struct stream_interface *si){	si->ib->flags &= ~CF_SHUTR_NOW;	if (si->ib->flags & CF_SHUTR)		return;	si->ib->flags |= CF_SHUTR;	si->ib->rex = TICK_ETERNITY;	si->flags &= ~SI_FL_WAIT_ROOM;	if (si->state != SI_ST_EST && si->state != SI_ST_CON)		return;	if (si->ob->flags & CF_SHUTW) {		si->state = SI_ST_DIS;		si->exp = TICK_ETERNITY;		si_applet_release(si);	}	else if (si->flags & SI_FL_NOHALF) {		/* we want to immediately forward this close to the write side */		return stream_int_shutw(si);	}	/* note that if the task exists, it must unregister itself once it runs */	if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)		task_wakeup(si->owner, TASK_WOKEN_IO);}

/* Register a function to handle a stream_interface as a standalone task. The * new task itself is returned and is assigned as si->owner. The stream_interface * pointer will be pointed to by the task's context. The handler can be detached * by using stream_int_unregister_handler(). * FIXME: the code should be updated to ensure that we don't change si->owner * anymore as this is not needed. However, process_session still relies on it. */struct task *stream_int_register_handler_task(struct stream_interface *si,					      struct task *(*fct)(struct task *)){	struct task *t;	DPRINTF(stderr, "registering handler %p for si %p (was %p)/n", fct, si, si->owner);	si->update  = stream_int_update;	si->shutr   = stream_int_shutr;	si->shutw   = stream_int_shutw;	si->chk_rcv = stream_int_chk_rcv;	si->chk_snd = stream_int_chk_snd;	si->connect = NULL;	clear_target(&si->target);	si->release   = NULL;	si->flags |= SI_FL_WAIT_DATA;	t = task_new();	si->owner = t;	if (!t)		return t;	set_target_task(&si->target, t);	t->process = fct;	t->context = si;	task_wakeup(si->owner, TASK_WOKEN_INIT);	return t;}

/* default update function for scheduled tasks, not used for embedded tasks */void stream_int_update(struct stream_interface *si){	DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x/n",		__FUNCTION__,		si, si->state, si->ib->flags, si->ob->flags);	if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)		task_wakeup(si->owner, TASK_WOKEN_IO);}

/** * Unblock the first task blocking on blocked_on */void task_unblock(void *blocked_on) {    struct task_struct *task = block_queue.tasks;    if(!task)        return;    for(;task != NULL; task = task->next_rq) {        if(blocked_on == task->blocked_on) {            /* It is the foreground, AND blocking on blocked_on */            task_wakeup(&block_queue, task);            /* Wake up ALL, do not break here. Consider blocking on pipes... */        }    }}

/** * Unblock the task which was waiting for the terminal. */void task_unblock_foreground(void *blocked_on) {    struct task_struct *task = block_queue.tasks;    if(!task)        return;    for(;task != NULL; task = task->next_rq) {        if(task->foreground && (blocked_on == task->blocked_on)) {            /* It is the foreground, AND blocking on blocked_on */            task_wakeup(&block_queue, task);            /* TODO: break; ???? */        }    }}

s32 sys_exit(){    u32 eflags;    _local_irq_save(eflags);    if (task_list[current].pwait != 0)         task_wakeup(task_list[current].pwait);    task_list[current].state = TASK_STOPED;    _local_irq_restore(eflags);    /* 等待0号任务清空 */    while(1){idle();};}

/* * Manages a server's connection queue. This function will try to dequeue as * many pending sessions as possible, and wake them up. */void process_srv_queue(struct server *s){    struct proxy  *p = s->proxy;    int maxconn;    /* First, check if we can handle some connections queued at the proxy. We     * will take as many as we can handle.     */    maxconn = srv_dynamic_maxconn(s);    while (s->served < maxconn) {        struct session *sess = pendconn_get_next_sess(s, p);        if (sess == NULL)            break;        task_wakeup(sess->task, TASK_WOKEN_RES);    }}

/* * Use this function to force a close of a peer session */static void peer_session_forceshutdown(struct session * session){	struct stream_interface *oldsi;	if (objt_applet(session->si[0].conn->target) == &peer_applet) {		oldsi = &session->si[0];	}	else {		oldsi = &session->si[1];	}	/* call release to reinit resync states if needed */	peer_session_release(oldsi);	oldsi->applet.st0 = PEER_SESSION_END;	oldsi->conn->xprt_ctx = NULL;	task_wakeup(session->task, TASK_WOKEN_MSG);}

/* * This function performs a shutdown-write on a stream interface attached to an * applet in a connected or init state (it does nothing for other states). It * either shuts the write side or marks itself as closed. The buffer flags are * updated to reflect the new state. It does also close everything if the SI * was marked as being in error state. The owner task is woken up if it exists. */static void stream_int_shutw(struct stream_interface *si){	struct channel *ic = si_ic(si);	struct channel *oc = si_oc(si);	oc->flags &= ~CF_SHUTW_NOW;	if (oc->flags & CF_SHUTW)		return;	oc->flags |= CF_SHUTW;	oc->wex = TICK_ETERNITY;	si->flags &= ~SI_FL_WAIT_DATA;	switch (si->state) {	case SI_ST_EST:		/* we have to shut before closing, otherwise some short messages		 * may never leave the system, especially when there are remaining		 * unread data in the socket input buffer, or when nolinger is set.		 * However, if SI_FL_NOLINGER is explicitly set, we know there is		 * no risk so we close both sides immediately.		 */		if (!(si->flags & (SI_FL_ERR | SI_FL_NOLINGER)) &&		    !(ic->flags & (CF_SHUTR|CF_DONT_READ)))			return;		/* fall through */	case SI_ST_CON:	case SI_ST_CER:	case SI_ST_QUE:	case SI_ST_TAR:		/* Note that none of these states may happen with applets */		si->state = SI_ST_DIS;		si_applet_release(si);	default:		si->flags &= ~(SI_FL_WAIT_ROOM | SI_FL_NOLINGER);		ic->flags &= ~CF_SHUTR_NOW;		ic->flags |= CF_SHUTR;		ic->rex = TICK_ETERNITY;		si->exp = TICK_ETERNITY;	}	/* note that if the task exists, it must unregister itself once it runs */	if (!(si->flags & SI_FL_DONT_WAKE))		task_wakeup(si_task(si), TASK_WOKEN_IO);}

/** * Free the task's kernel stack, memory context, and close open files. * NOTE: Only called when removed from its queue. * Leaves the task on in the list of all tasks (only remove when a * parent calls cleanup_child). */void task_destroy(struct task_struct *task) {    int i;    mm_destroy(task->mm);    free_page(ALIGN_DOWN(task->kernel_rsp, PAGE_SIZE));    /* Close any open files */    for(i = 0; i < TASK_FILES_MAX; i++) {        struct file *fp = task->files[i];        if(fp) {            fp->f_op->close(fp);            task->files[i] = NULL;        }    }    /* Give terminal control back to the parent */    if(task->foreground && task->parent) {        task->parent->foreground = 1;    }    task->foreground = 0; /* remove the foreground from the dead task */    /* If we have children give them to init */    if(task->chld) {        /* Give all children to init */        struct task_struct *prevchld, *nextchld;        for(prevchld = task->chld; prevchld != NULL; prevchld = nextchld) {            nextchld = prevchld->sib;            add_child(init_task, prevchld);        }    }    if(task->parent) {        /* Notify parent of child's termination */        if(task->parent->state == TASK_WAITING) {            task_wakeup(&wait_queue, task->parent);        }    } else {        /* We have no parent so add ourself to init */        add_child(init_task, task);    }}

/* default chk_rcv function for scheduled tasks */static void stream_int_chk_rcv(struct stream_interface *si){	struct channel *ib = si->ib;	DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x/n",		__FUNCTION__,		si, si->state, si->ib->flags, si->ob->flags);	if (unlikely(si->state != SI_ST_EST || (ib->flags & (CF_SHUTR|CF_DONT_READ))))		return;	if (channel_full(ib)) {		/* stop reading */		si->flags |= SI_FL_WAIT_ROOM;	}	else {		/* (re)start reading */		si->flags &= ~SI_FL_WAIT_ROOM;		if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)			task_wakeup(si->owner, TASK_WOKEN_IO);	}}

/* default chk_rcv function for scheduled tasks */static void stream_int_chk_rcv(struct stream_interface *si){	struct channel *ic = si_ic(si);	DPRINTF(stderr, "%s: si=%p, si->state=%d ic->flags=%08x oc->flags=%08x/n",		__FUNCTION__,		si, si->state, ic->flags, si_oc(si)->flags);	if (unlikely(si->state != SI_ST_EST || (ic->flags & (CF_SHUTR|CF_DONT_READ))))		return;	if (!channel_may_recv(ic) || ic->pipe) {		/* stop reading */		si->flags |= SI_FL_WAIT_ROOM;	}	else {		/* (re)start reading */		si->flags &= ~SI_FL_WAIT_ROOM;		if (!(si->flags & SI_FL_DONT_WAKE))			task_wakeup(si_task(si), TASK_WOKEN_IO);	}}

/* * This function performs a shutdown-read on a stream interface in a connected * or init state (it does nothing for other states). It either shuts the read * side or marks itself as closed. The buffer flags are updated to reflect the * new state. If the stream interface has SI_FL_NOHALF, we also forward the * close to the write side. If a control layer is defined, then it is supposed * to be a socket layer and file descriptors are then shutdown or closed * accordingly. If no control layer is defined, then the SI is supposed to be * an embedded one and the owner task is woken up if it exists. The function * does not disable polling on the FD by itself, it returns non-zero instead * if the caller needs to do so (except when the FD is deleted where this is * implicit). */int stream_int_shutr(struct stream_interface *si){	struct connection *conn = si->conn;	si->ib->flags &= ~CF_SHUTR_NOW;	if (si->ib->flags & CF_SHUTR)		return 0;	si->ib->flags |= CF_SHUTR;	si->ib->rex = TICK_ETERNITY;	si->flags &= ~SI_FL_WAIT_ROOM;	if (si->state != SI_ST_EST && si->state != SI_ST_CON)		return 0;	if (si->ob->flags & CF_SHUTW) {		conn_full_close(si->conn);		si->state = SI_ST_DIS;		si->exp = TICK_ETERNITY;		if (si->release)			si->release(si);	}	else if (si->flags & SI_FL_NOHALF) {		/* we want to immediately forward this close to the write side */		return stream_int_shutw(si);	}	else if (conn->ctrl) {		/* we want the caller to disable polling on this FD */		return 1;	}	/* note that if the task exists, it must unregister itself once it runs */	if (!conn->ctrl && !(si->flags & SI_FL_DONT_WAKE) && si->owner)		task_wakeup(si->owner, TASK_WOKEN_IO);	return 0;}

/* Redistribute pending connections when a server goes down. The number of * connections redistributed is returned. It must be called with the server * lock held. */int pendconn_redistribute(struct server *s){	struct pendconn *p;	struct eb32_node *node;	int xferred = 0;	/* The REDISP option was specified. We will ignore cookie and force to	 * balance or use the dispatcher. */	if ((s->proxy->options & (PR_O_REDISP|PR_O_PERSIST)) != PR_O_REDISP)		return 0;	for (node = eb32_first(&s->pendconns); node; node = eb32_next(node)) {		p = eb32_entry(&node, struct pendconn, node);		if (p->strm_flags & SF_FORCE_PRST)			continue;		/* it's left to the dispatcher to choose a server */		__pendconn_unlink(p);		p->strm_flags &= ~(SF_DIRECT | SF_ASSIGNED | SF_ADDR_SET);		task_wakeup(p->strm->task, TASK_WOKEN_RES);	}	return xferred;}

/* default update function for embedded tasks, to be used at the end of the i/o handler */void stream_int_update_embedded(struct stream_interface *si){	int old_flags = si->flags;	DPRINTF(stderr, "%s: si=%p, si->state=%d ib->flags=%08x ob->flags=%08x/n",		__FUNCTION__,		si, si->state, si->ib->flags, si->ob->flags);	if (si->state != SI_ST_EST)		return;	if ((si->ob->flags & (BF_OUT_EMPTY|BF_SHUTW|BF_HIJACK|BF_SHUTW_NOW)) == (BF_OUT_EMPTY|BF_SHUTW_NOW))		si->shutw(si);	if ((si->ob->flags & (BF_FULL|BF_SHUTW|BF_SHUTW_NOW|BF_HIJACK)) == 0)		si->flags |= SI_FL_WAIT_DATA;	/* we're almost sure that we need some space if the buffer is not	 * empty, even if it's not full, because the applets can't fill it.	 */	if ((si->ib->flags & (BF_SHUTR|BF_OUT_EMPTY|BF_DONT_READ)) == 0)		si->flags |= SI_FL_WAIT_ROOM;	if (si->ob->flags & BF_WRITE_ACTIVITY) {		if (tick_isset(si->ob->wex))			si->ob->wex = tick_add_ifset(now_ms, si->ob->wto);	}	if (si->ib->flags & BF_READ_ACTIVITY ||	    (si->ob->flags & BF_WRITE_ACTIVITY && !(si->flags & SI_FL_INDEP_STR))) {		if (tick_isset(si->ib->rex))			si->ib->rex = tick_add_ifset(now_ms, si->ib->rto);	}	/* save flags to detect changes */	old_flags = si->flags;	if (likely((si->ob->flags & (BF_SHUTW|BF_WRITE_PARTIAL|BF_FULL|BF_DONT_READ)) == BF_WRITE_PARTIAL &&		   (si->ob->prod->flags & SI_FL_WAIT_ROOM)))		si->ob->prod->chk_rcv(si->ob->prod);	if (((si->ib->flags & (BF_READ_PARTIAL|BF_OUT_EMPTY)) == BF_READ_PARTIAL) &&	    (si->ib->cons->flags & SI_FL_WAIT_DATA)) {		si->ib->cons->chk_snd(si->ib->cons);		/* check if the consumer has freed some space */		if (!(si->ib->flags & BF_FULL))			si->flags &= ~SI_FL_WAIT_ROOM;	}	/* Note that we're trying to wake up in two conditions here :	 *  - special event, which needs the holder task attention	 *  - status indicating that the applet can go on working. This	 *    is rather hard because we might be blocking on output and	 *    don't want to wake up on input and vice-versa. The idea is	 *    to only rely on the changes the chk_* might have performed.	 */	if (/* check stream interface changes */	    ((old_flags & ~si->flags) & (SI_FL_WAIT_ROOM|SI_FL_WAIT_DATA)) ||	    /* changes on the production side */	    (si->ib->flags & (BF_READ_NULL|BF_READ_ERROR)) ||	    si->state != SI_ST_EST ||	    (si->flags & SI_FL_ERR) ||	    ((si->ib->flags & BF_READ_PARTIAL) &&	     (!si->ib->to_forward || si->ib->cons->state != SI_ST_EST)) ||	    /* changes on the consumption side */	    (si->ob->flags & (BF_WRITE_NULL|BF_WRITE_ERROR)) ||	    ((si->ob->flags & BF_WRITE_ACTIVITY) &&	     ((si->ob->flags & BF_SHUTW) ||	      si->ob->prod->state != SI_ST_EST ||	      ((si->ob->flags & BF_OUT_EMPTY) && !si->ob->to_forward)))) {		if (!(si->flags & SI_FL_DONT_WAKE) && si->owner)			task_wakeup(si->owner, TASK_WOKEN_IO);	}	if (si->ib->flags & BF_READ_ACTIVITY)		si->ib->flags &= ~BF_READ_DONTWAIT;}

/* * Task processing function to manage re-connect and peer session * tasks wakeup on local update. */static struct task *process_peer_sync(struct task * task){	struct shared_table *st = (struct shared_table *)task->context;	struct peer_session *ps;	task->expire = TICK_ETERNITY;	if (!stopping) {		/* Normal case (not soft stop)*/		if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL) &&		     (!nb_oldpids || tick_is_expired(st->resync_timeout, now_ms)) &&		     !(st->flags & SHTABLE_F_RESYNC_ASSIGN)) {			/* Resync from local peer needed			   no peer was assigned for the lesson			   and no old local peer found			       or resync timeout expire */			/* flag no more resync from local, to try resync from remotes */			st->flags |= SHTABLE_F_RESYNC_LOCAL;			/* reschedule a resync */			st->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));		}		/* For each session */		for (ps = st->sessions; ps; ps = ps->next) {			/* For each remote peers */			if (!ps->peer->local) {				if (!ps->session) {					/* no active session */					if (ps->statuscode == 0 ||					    ps->statuscode == PEER_SESSION_SUCCESSCODE ||					    ((ps->statuscode == PEER_SESSION_CONNECTCODE ||					      ps->statuscode == PEER_SESSION_CONNECTEDCODE) &&					     tick_is_expired(ps->reconnect, now_ms))) {						/* connection never tried						 * or previous session established with success						 * or previous session failed during connection						 * and reconnection timer is expired */						/* retry a connect */						ps->session = peer_session_create(ps->peer, ps);					}					else if (ps->statuscode == PEER_SESSION_CONNECTCODE ||						 ps->statuscode == PEER_SESSION_CONNECTEDCODE) {						/* If previous session failed during connection						 * but reconnection timer is not expired */						/* reschedule task for reconnect */						task->expire = tick_first(task->expire, ps->reconnect);					}					/* else do nothing */				} /* !ps->session */				else if (ps->statuscode == PEER_SESSION_SUCCESSCODE) {					/* current session is active and established */					if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&					    !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&					    !(ps->flags & PEER_F_LEARN_NOTUP2DATE)) {						/* Resync from a remote is needed						 * and no peer was assigned for lesson						 * and current peer may be up2date */						/* assign peer for the lesson */						ps->flags |= PEER_F_LEARN_ASSIGN;						st->flags |= SHTABLE_F_RESYNC_ASSIGN;						/* awake peer session task to handle a request of resync */						task_wakeup(ps->session->task, TASK_WOKEN_MSG);					}					else if ((int)(ps->pushed - ps->table->table->localupdate) < 0) {						/* awake peer session task to push local updates */						task_wakeup(ps->session->task, TASK_WOKEN_MSG);					}					/* else do nothing */				} /* SUCCESSCODE */			} /* !ps->peer->local */		} /* for */		/* Resync from remotes expired: consider resync is finished */		if (((st->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE) &&		    !(st->flags & SHTABLE_F_RESYNC_ASSIGN) &&		    tick_is_expired(st->resync_timeout, now_ms)) {			/* Resync from remote peer needed			 * no peer was assigned for the lesson			 * and resync timeout expire */			/* flag no more resync from remote, consider resync is finished */			st->flags |= SHTABLE_F_RESYNC_REMOTE;		}		if ((st->flags & SHTABLE_RESYNC_STATEMASK) != SHTABLE_RESYNC_FINISHED) {			/* Resync not finished*/			/* reschedule task to resync timeout, to ended resync if needed */			task->expire = tick_first(task->expire, st->resync_timeout);		}	} /* !stopping *///.........这里部分代码省略.........

//.........这里部分代码省略.........				if (!st){					si->applet.st0 = PEER_SESSION_EXIT;					si->applet.st1 = PEER_SESSION_ERRTABLE;					goto switchstate;				}				/* If no peer session for current peer */				if (!ps) {					si->applet.st0 = PEER_SESSION_EXIT;					si->applet.st1 = PEER_SESSION_ERRPEER;					goto switchstate;				}				si->conn.data_ctx = ps;				si->applet.st0 = PEER_SESSION_SENDSUCCESS;				/* fall through */			}			case PEER_SESSION_SENDSUCCESS:{				struct peer_session *ps = (struct peer_session *)si->conn.data_ctx;				repl = snprintf(trash, trashlen, "%d/n", PEER_SESSION_SUCCESSCODE);				repl = bi_putblk(si->ib, trash, repl);				if (repl <= 0) {					if (repl == -1)						goto out;					si->applet.st0 = PEER_SESSION_END;					goto switchstate;				}				/* Register status code */				ps->statuscode = PEER_SESSION_SUCCESSCODE;				/* Awake main task */				task_wakeup(ps->table->sync_task, TASK_WOKEN_MSG);				/* Init cursors */				ps->teaching_origin =ps->lastpush = ps->lastack = ps->pushack = 0;				ps->pushed = ps->update;				/* Init confirm counter */				ps->confirm = 0;				/* reset teaching and learning flags to 0 */				ps->flags &= PEER_TEACH_RESET;				ps->flags &= PEER_LEARN_RESET;				/* if current peer is local */				if (ps->peer->local) {					/* if table need resyncfrom local and no process assined  */					if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMLOCAL &&					    !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {						/* assign local peer for a lesson, consider lesson already requested */						ps->flags |= PEER_F_LEARN_ASSIGN;						ps->table->flags |= (SHTABLE_F_RESYNC_ASSIGN|SHTABLE_F_RESYNC_PROCESS);					}				}				else if ((ps->table->flags & SHTABLE_RESYNC_STATEMASK) == SHTABLE_RESYNC_FROMREMOTE &&					 !(ps->table->flags & SHTABLE_F_RESYNC_ASSIGN)) {					/* assign peer for a lesson  */					ps->flags |= PEER_F_LEARN_ASSIGN;					ps->table->flags |= SHTABLE_F_RESYNC_ASSIGN;				}				/* switch to waiting message state */				si->applet.st0 = PEER_SESSION_WAITMSG;				goto switchstate;