/[hurd]/hurd/pflocal/connq.c
ViewVC logotype

Diff of /hurd/pflocal/connq.c

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.16 by neal, Mon Aug 29 08:29:35 2005 UTC revision 1.17 by neal, Mon Aug 29 09:41:21 2005 UTC
# Line 1  Line 1 
1  /* Listen queue functions  /* Listen queue functions
2    
3     Copyright (C) 1995,96,2001, 2005 Free Software Foundation, Inc.     Copyright (C) 1995,96,2001 Free Software Foundation, Inc.
4    
5     Written by Miles Bader <miles@gnu.org>     Written by Miles Bader <miles@gnu.org>
6    
# Line 26  Line 26 
26  /* A queue for queueing incoming connections.  */  /* A queue for queueing incoming connections.  */
27  struct connq  struct connq
28  {  {
29      /* True if all connection requests should be treated as non-blocking.  */
30      int noqueue;
31    
32    /* The connection request queue.  */    /* The connection request queue.  */
33    struct connq_request *head;    struct connq_request **queue;
34    struct connq_request **tail;    unsigned length;
35    unsigned count;    /* Head is the position in QUEUE of the first request, and TAIL is the
36    unsigned max;       first free position in the queue.  If HEAD == TAIL, then the queue is
37         empty.  Starting at HEAD, successive positions can be calculated by
38         using qnext().  */
39      unsigned head, tail;
40    
41    /* Threads that have done an accept on this queue wait on this condition.  */    /* Threads that have done an accept on this queue wait on this condition.  */
42    struct condition listeners;    struct condition listeners;
43    unsigned num_listeners;    unsigned num_listeners;
44    
   /* Threads that have done a connect on this queue wait on this condition.  */  
   struct condition connectors;  
   unsigned num_connectors;  
   
45    struct mutex lock;    struct mutex lock;
46  };  };
47    
48    /* Returns the position CQ's queue after POS.  */
49    static inline unsigned
50    qnext (struct connq *cq, unsigned pos)
51    {
52      return (pos + 1 == cq->length) ? 0 : pos + 1;
53    }
54    
55  /* ---------------------------------------------------------------- */  /* ---------------------------------------------------------------- */
56    
# Line 49  struct connq Line 58  struct connq
58     get information from and to the thread.  */     get information from and to the thread.  */
59  struct connq_request  struct connq_request
60  {  {
   struct connq_request *next;  
   
61    /* The socket that's waiting to connect.  */    /* The socket that's waiting to connect.  */
62    struct sock *sock;    struct sock *sock;
63    
64      /* What the waiting thread blocks on.  */
65      struct condition signal;
66      struct mutex lock;
67    
68      /* Set to true when this request has been dealt with, to guard against
69         spurious conditions being signaled.  */
70      int completed;
71    
72      /* After the waiting thread is unblocked, this is the result, either 0 if
73         SOCK has been connected, or an error.  */
74      error_t err;
75  };  };
76    
77  static inline void  static inline void
78  connq_request_init (struct connq_request *req, struct sock *sock)  connq_request_init (struct connq_request *req, struct sock *sock)
79  {  {
80      req->err = 0;
81    req->sock = sock;    req->sock = sock;
82  }    req->completed = 0;
83      condition_init (&req->signal);
84  /* Enqueue connection request REQ onto CQ.  CQ must be locked.  */    mutex_init (&req->lock);
 static void  
 connq_request_enqueue (struct connq *cq, struct connq_request *req)  
 {  
   assert (! mutex_try_lock (&cq->lock));  
   
   req->next = NULL;  
   *cq->tail = req;  
   cq->tail = &req->next;  
   
   cq->count ++;  
 }  
   
 /* Dequeue a pending request from CQ.  CQ must be locked and must not  
    be empty.  */  
 static struct connq_request *  
 connq_request_dequeue (struct connq *cq)  
 {  
   struct connq_request *req;  
   
   assert (! mutex_try_lock (&cq->lock));  
   assert (cq->head);  
   
   req = cq->head;  
   cq->head = req->next;  
   if (! cq->head)  
     /* We just dequeued the last element.  Fixup the tail pointer.  */  
     cq->tail = &cq->head;  
   
   cq->count --;  
   
   return req;  
85  }  }
86    
87  /* ---------------------------------------------------------------- */  /* ---------------------------------------------------------------- */
# Line 106  connq_create (struct connq **cq) Line 95  connq_create (struct connq **cq)
95    struct connq *new = malloc (sizeof (struct connq));    struct connq *new = malloc (sizeof (struct connq));
96    
97    if (!new)    if (!new)
98      return ENOBUFS;      return ENOMEM;
   
   new->head = NULL;  
   new->tail = &new->head;  
   new->count = 0;  
   /* By default, don't queue requests.  */  
   new->max = 0;  
99    
100      new->noqueue = 1;             /* By default, don't queue requests.  */
101      new->length = 0;
102      new->head = new->tail = 0;
103      new->queue = NULL;
104    new->num_listeners = 0;    new->num_listeners = 0;
   new->num_connectors = 0;  
105    
106    mutex_init (&new->lock);    mutex_init (&new->lock);
107    condition_init (&new->listeners);    condition_init (&new->listeners);
   condition_init (&new->connectors);  
108    
109    *cq = new;    *cq = new;
110    return 0;    return 0;
# Line 131  connq_destroy (struct connq *cq) Line 116  connq_destroy (struct connq *cq)
116  {  {
117    /* Everybody in the queue should hold a reference to the socket    /* Everybody in the queue should hold a reference to the socket
118       containing the queue.  */       containing the queue.  */
119    assert (! cq->head);    assert (cq->length == 0);
120    assert (cq->count == 0);    /* Nevertheless, malloc(0) or realloc(0) might allocate some small
121         space.  */
122      if (cq->queue)
123        free (cq->queue);
124    free (cq);    free (cq);
125  }  }
126    
127  /* ---------------------------------------------------------------- */  /* ---------------------------------------------------------------- */
128    
129  /* Return a connection request on CQ.  If SOCK is NULL, the request is  /* Wait for a connection attempt to be made on CQ, and return the connecting
130     left in the queue.  If NOBLOCK is true, EWOULDBLOCK is returned     socket in SOCK, and a request tag in REQ.  If REQ is NULL, the request is
131     when there are no immediate connections available.  */     left in the queue, otherwise connq_request_complete must be called on REQ
132       to allow the requesting thread to continue.  If NOBLOCK is true,
133       EWOULDBLOCK is returned when there are no immediate connections
134       available. */
135  error_t  error_t
136  connq_listen (struct connq *cq, int noblock, struct sock **sock)  connq_listen (struct connq *cq, int noblock,
137                  struct connq_request **req, struct sock **sock)
138  {  {
   error_t err = 0;  
   
139    mutex_lock (&cq->lock);    mutex_lock (&cq->lock);
140    
141    if (noblock && cq->count == 0 && cq->num_connectors == 0)    if (noblock && cq->head == cq->tail)
142      {      {
143        mutex_unlock (&cq->lock);        mutex_unlock (&cq->lock);
144        return EWOULDBLOCK;        return EWOULDBLOCK;
145      }      }
146    
   if (! sock && (cq->count > 0 || cq->num_connectors > 0))  
     /* The caller just wants to know if a connection ready.  */  
     {  
       mutex_unlock (&cq->lock);  
       return 0;  
     }  
   
147    cq->num_listeners++;    cq->num_listeners++;
148    
149    if (cq->count == 0)    while (cq->head == cq->tail)
150      /* The request queue is empty.  */      if (hurd_condition_wait (&cq->listeners, &cq->lock))
151      {        {
152        assert (! cq->head);          cq->num_listeners--;
153            mutex_unlock (&cq->lock);
154        if (cq->num_connectors > 0)          return EINTR;
155          /* Someone is waiting for an acceptor.  Signal that we can        }
            service their request.  */  
         condition_signal (&cq->connectors);  
   
       do  
         if (hurd_condition_wait (&cq->listeners, &cq->lock))  
           {  
             cq->num_listeners--;  
             err = EINTR;  
             goto out;  
           }  
       while (cq->count == 0);  
     }  
   
   assert (cq->head);  
156    
157    if (sock)    if (req != NULL)
158      /* Dequeue the next request, if desired.  */      /* Dequeue the next request, if desired.  */
159      {      {
160        struct connq_request *req = connq_request_dequeue (cq);        *req = cq->queue[cq->head];
161        *sock = req->sock;        cq->head = qnext (cq, cq->head);
162        free (req);        if (sock != NULL)
163            *sock = (*req)->sock;
164      }      }
   else if (cq->num_listeners > 0)  
     /* The caller will not actually process this request but someone  
        else could.  (This case is rare but possible: it would require  
        one thread to do a select on the socket and a second to do an  
        accept.)  */  
     condition_signal (&cq->listeners);  
   else  
     /* There is no one else to process the request and the connection  
        has now been initiated.  This is not actually a problem as even  
        if the current queue limit is 0, the connector will queue the  
        request and another listener (should) eventually come along.  
        (In fact it is very probably as the caller has likely done a  
        select and will now follow up with an accept.)  */  
     ;  
165    
166   out:    cq->num_listeners--;
167    
168    mutex_unlock (&cq->lock);    mutex_unlock (&cq->lock);
169    return err;  
170      return 0;
171    }
172    
173    /* Return the error code ERR to the thread that made the listen request REQ,
174       returned from a previous connq_listen.  */
175    void
176    connq_request_complete (struct connq_request *req, error_t err)
177    {
178      mutex_lock (&req->lock);
179      req->err = err;
180      req->completed = 1;
181      condition_signal (&req->signal);
182      mutex_unlock (&req->lock);
183  }  }
184    
185  /* Try to connect SOCK with the socket listening on CQ.  If NOBLOCK is  /* Try to connect SOCK with the socket listening on CQ.  If NOBLOCK is true,
186     true, then return EWOULDBLOCK if there are no connections     then return EWOULDBLOCK immediately when there are no immediate
187     immediately available.  On success, this call must be followed up     connections available.  Neither SOCK nor CQ should be locked.  */
    either connq_connect_complete or connq_connect_cancel.  */  
188  error_t  error_t
189  connq_connect (struct connq *cq, int noblock)  connq_connect (struct connq *cq, int noblock, struct sock *sock)
190  {  {
191      error_t err = 0;
192      unsigned next;
193    
194    mutex_lock (&cq->lock);    mutex_lock (&cq->lock);
195    
196    /* Check for listeners after we've locked CQ for good.  */    /* Check for listeners after we've locked CQ for good.  */
197      if ((noblock || cq->noqueue) && cq->num_listeners == 0)
   if (noblock  
       && cq->count + cq->num_connectors >= cq->max + cq->num_listeners)  
     /* We are in non-blocking mode and would have to wait to secure an  
        entry in the listen queue.  */  
198      {      {
199        mutex_unlock (&cq->lock);        mutex_unlock (&cq->lock);
200        return EWOULDBLOCK;        return EWOULDBLOCK;
201      }      }
202    
203    cq->num_connectors ++;    next = qnext (cq, cq->tail);
204      if (next == cq->tail)
205    while (cq->count + cq->num_connectors > cq->max + cq->num_listeners)      /* The queue is full.  */
206      /* The queue is full and there is no immediate listener to service      err = ECONNREFUSED;
207         us.  Block until we can get a slot.  */    else
208      if (hurd_condition_wait (&cq->connectors, &cq->lock))      {
209        {        struct connq_request req;
         cq->num_connectors --;  
         mutex_unlock (&cq->lock);  
         return EINTR;  
       }  
   
   mutex_unlock (&cq->lock);  
   
   return 0;  
 }  
   
 /* Follow up to connq_connect.  Completes the connect, SOCK is the new  
    server socket.  */  
 void  
 connq_connect_complete (struct connq *cq, struct sock *sock)  
 {  
   struct connq_request *req;  
   
   req = malloc (sizeof (struct connq_request));  
   if (! req)  
     abort ();  
210    
211    connq_request_init (req, sock);        connq_request_init (&req, sock);
212    
213    mutex_lock (&cq->lock);        cq->queue[cq->tail] = &req;
214          cq->tail = next;
215    
216    assert (cq->num_connectors > 0);        /* Hold REQ.LOCK before we signal the condition so that we're sure
217    cq->num_connectors --;           to be woken up.  */
218          mutex_lock (&req.lock);
219          condition_signal (&cq->listeners);
220          mutex_unlock (&cq->lock);
221    
222    connq_request_enqueue (cq, req);        while (!req.completed)
223            condition_wait (&req.signal, &req.lock);
224          err = req.err;
225    
226    if (cq->num_listeners > 0)        mutex_unlock (&req.lock);
     /* Wake a listener up.  We must consume the listener ref here as  
        someone else might call this function before the listener  
        thread dequeues this request.  */  
     {  
       cq->num_listeners --;  
       condition_signal (&cq->listeners);  
227      }      }
228    
229    mutex_unlock (&cq->lock);    return err;
230  }  }
231    
232  /* Follow up to connq_connect.  Cancel the connect.  */  #if 0
233  void  /* `Compresses' CQ, by removing any NULL entries.  CQ should be locked.  */
234  connq_connect_cancel (struct connq *cq)  static void
235    connq_compress (struct connq *cq)
236  {  {
237    mutex_lock (&cq->lock);    unsigned pos;
238      unsigned comp_tail = cq->head;
   assert (cq->num_connectors > 0);  
   cq->num_connectors --;  
239    
240    if (cq->count + cq->num_connectors >= cq->max + cq->num_listeners)    /* Now compress the queue to remove any null entries we put in.  */
241      /* A connector is blocked and could use the spot we reserved.  */    for (pos = cq->head; pos != cq->tail; pos = qnext (cq, pos))
242      condition_signal (&cq->connectors);      if (cq->queue[pos] != NULL)
243          /* This position has a non-NULL request, so move it to the end of the
244             compressed queue.  */
245          {
246            cq->queue[comp_tail] = cq->queue[pos];
247            comp_tail = qnext (cq, comp_tail);
248          }
249    
250    mutex_unlock (&cq->lock);    /* Move back tail to only include what we kept in the queue.  */
251      cq->tail = comp_tail;
252  }  }
253    #endif
254    
255  /* Set CQ's queue length to LENGTH.  */  /* Set CQ's queue length to LENGTH.  Any sockets already waiting for a
256       connections that are past the new length will fail with ECONNREFUSED.  */
257  error_t  error_t
258  connq_set_length (struct connq *cq, int max)  connq_set_length (struct connq *cq, int length)
259  {  {
   int omax;  
   
260    mutex_lock (&cq->lock);    mutex_lock (&cq->lock);
   omax = cq->max;  
   cq->max = max;  
261    
262    if (max > omax && cq->count >= omax && cq->count < max    if (length > cq->length)
263        && cq->num_connectors >= cq->num_listeners)      /* Growing the queue is simple... */
264      /* This is an increase in the number of connection slots which has      cq->queue = realloc (cq->queue, sizeof (struct connq_request *) * length);
265         made some slots available and there are waiting threads.  Wake    else
266         them up.  */      /* Shrinking it less so.  */
267      condition_broadcast (&cq->listeners);      {
268          int i;
269          struct connq_request **new_queue =
270            malloc (sizeof (struct connq_request *) * length);
271    
272          for (i = 0; i < cq->length && cq->head != cq->tail; i++)
273            {
274              if (i < length)
275                /* Keep this connect request in the queue.  */
276                new_queue[length - i] = cq->queue[cq->head];
277              else
278                /* Punt this one.  */
279                connq_request_complete (cq->queue[cq->head], ECONNREFUSED);
280              cq->head = qnext (cq, cq->head);
281            }
282    
283          free (cq->queue);
284          cq->queue = new_queue;
285        }
286    
287      cq->noqueue = 0;              /* Turn on queueing.  */
288    
289    mutex_unlock (&cq->lock);    mutex_unlock (&cq->lock);
290    

Legend:
Removed from v.1.16  
changed lines
  Added in v.1.17

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26