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

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

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