[sheepdog] [PATCH 1/2] sheep: introduce async request handling

[Liu Yuan]

On Tue, Dec 17, 2013 at 03:13:26PM +0900, Hitoshi Mitake wrote:
> At Mon, 16 Dec 2013 21:16:38 +0800,
> Liu Yuan wrote:
> > 
> > This patch introduces async request framework that has the following usage
> > patern:
> > 
> > iocb = local_req_init();
> > 
> > loop:
> >     ...
> >     exec_local_req_async(req, iocb);
> >     ...
> > 
> > ret = local_req_wait(iocb);
> > 
> > wait until all the reqs finish, if any req fails, just return it from
> > local_req_wait();
> > 
> > Request throttling should be handled by the async user.
> > 
> > Signed-off-by: Liu Yuan <namei.unix at gmail.com>
> > ---
> >  sheep/request.c    |   85 ++++++++++++++++++++++++++++++++++++++++++++++++----
> >  sheep/sheep.c      |    2 +-
> >  sheep/sheep_priv.h |   13 +++++++-
> >  3 files changed, 92 insertions(+), 8 deletions(-)
> 
> Just a quick review.
> 
> > 
> > diff --git a/sheep/request.c b/sheep/request.c
> > index dfca8ee..53ccc04 100644
> > --- a/sheep/request.c
> > +++ b/sheep/request.c
> > @@ -499,6 +499,15 @@ static void free_local_request(struct request *req)
> >  	free(req);
> >  }
> >  
> > +static void submit_local_request(struct request *req)
> > +{
> > +	pthread_mutex_lock(&sys->local_req_lock);
> > +	list_add_tail(&req->request_list, &sys->local_req_queue);
> > +	pthread_mutex_unlock(&sys->local_req_lock);
> > +
> > +	eventfd_xwrite(sys->local_req_efd, 1);
> > +}
> > +
> 
> It seems that sys->local_req_efd is initialized with the return value
> of eventfd(0, 0); IIUC, event write to a fd created by eventfd(0, 0)
> can be blocked. SHould we use eventfd(0, EFD_NONBLOCK); for
> asynchronous requests?

sys->local_req_efd = eventfd(0, EFD_NONBLOCK); # line 912

> 
> >  /*
> >   * Exec the request locally and synchronously.
> >   *
> > @@ -514,16 +523,13 @@ worker_fn int exec_local_req(struct sd_req *rq, void *data)
> >  	req->rq = *rq;
> >  	req->local_req_efd = eventfd(0, 0);
> >  	if (req->local_req_efd < 0) {
> > +		sd_err("eventfd failed, %m");
> >  		/* Fake the result to ask for retry */
> >  		req->rp.result = SD_RES_NETWORK_ERROR;
> >  		goto out;
> >  	}
> >  
> > -	pthread_mutex_lock(&sys->local_req_lock);
> > -	list_add_tail(&req->request_list, &sys->local_req_queue);
> > -	pthread_mutex_unlock(&sys->local_req_lock);
> > -
> > -	eventfd_xwrite(sys->local_req_efd, 1);
> > +	submit_local_request(req);
> >  	eventfd_xread(req->local_req_efd);
> >  out:
> >  	/* fill rq with response header as exec_req does */
> > @@ -536,6 +542,73 @@ out:
> >  	return ret;
> >  }
> >  
> > +static void local_req_async_handler(int fd, int events, void *data)
> > +{
> > +	struct request *req = data;
> > +	struct request_iocb *iocb = req->iocb;
> > +
> > +	if (events & EPOLLERR)
> > +		sd_err("request handler error");
> > +	eventfd_xread(fd);
> > +
> > +	if (unlikely(req->rp.result != SD_RES_SUCCESS))
> > +		iocb->result = req->rp.result;
> > +
> > +	if (uatomic_sub_return(&iocb->count, 1) == 0)
> > +		eventfd_xwrite(iocb->efd, 1);
> > +
> > +	unregister_event(req->local_req_efd);
> > +	close(req->local_req_efd);
> > +	free_local_request(req);
> > +}
> > +
> > +worker_fn struct request_iocb *local_req_init(void)
> > +{
> > +	struct request_iocb *iocb = xzalloc(sizeof(*iocb));
> > +
> > +	iocb->efd = eventfd(0, 0);
> > +	if (iocb->efd < 0) {
> > +		sd_err("eventfd failed, %m");
> > +		free(iocb);
> > +		return NULL;
> > +	}
> > +	iocb->result = SD_RES_SUCCESS;
> > +	return iocb;
> > +}
> > +
> > +worker_fn int local_req_wait(struct request_iocb *iocb)
> > +{
> > +	int ret;
> > +
> > +	eventfd_xread(iocb->efd);
> > +
> > +	ret = iocb->result;
> > +	close(iocb->efd);
> > +	free(iocb);
> > +	return ret;
> > +}
> > +
> > +worker_fn int exec_local_req_async(struct sd_req *rq, void *data,
> > +				   struct request_iocb *iocb)
> > +{
> > +	struct request *req;
> > +
> > +	req = alloc_local_request(data, rq->data_length);
> > +	req->rq = *rq;
> > +	req->local_req_efd = eventfd(0, EFD_NONBLOCK);
> > +	if (req->local_req_efd < 0) {
> > +		sd_err("eventfd failed, %m");
> > +		return SD_RES_SYSTEM_ERROR;
> > +	}
> > +
> > +	uatomic_inc(&iocb->count);
> > +	req->iocb = iocb;
> > +	register_event(req->local_req_efd, local_req_async_handler,
> > req);
> 
> We cannot call register_event() in worker threads. It modifies data
> structures of the main event loop.
> 

what kind of data structure? i notice epoll_ctl is thread safe.

Thanks
Yuan