libsoup/soup-message-queue.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-message-queue.c: Message queue
   4  *
   5  * Copyright (C) 2003 Novell, Inc.
   6  * Copyright (C) 2008 Red Hat, Inc.
   7  */
   8
   9 #ifdef HAVE_CONFIG_H
  10 #include <config.h>
  11 #endif
  12
  13 #include "soup-message-queue.h"
  14 #include "soup.h"
  15
  16 /* This is an internal structure used by #SoupSession and its
  17  * subclasses to keep track of the status of messages currently being
  18  * processed.
  19  *
  20  * The #SoupMessageQueue itself is mostly just a linked list of
  21  * #SoupMessageQueueItem, with some added cleverness to allow the list
  22  * to be walked safely while other threads / re-entrant loops are
  23  * adding items to and removing items from it. In particular, this is
  24  * handled by refcounting items and then keeping "removed" items in
  25  * the list until their ref_count drops to 0, but skipping over the
  26  * "removed" ones when walking the queue.
  27  **/
  28
  29 struct _SoupMessageQueue {
  30         SoupSession *session;
  31
  32         GMutex mutex;
  33         SoupMessageQueueItem *head, *tail;
  34 };
  35
  36 SoupMessageQueue *
  37 soup_message_queue_new (SoupSession *session)
  38 {
  39         SoupMessageQueue *queue;
  40
  41         queue = g_slice_new0 (SoupMessageQueue);
  42         queue->session = session;
  43         g_mutex_init (&queue->mutex);
  44         return queue;
  45 }
  46
  47 void
  48 soup_message_queue_destroy (SoupMessageQueue *queue)
  49 {
  50         g_return_if_fail (queue->head == NULL);
  51
  52         g_mutex_clear (&queue->mutex);
  53         g_slice_free (SoupMessageQueue, queue);
  54 }
  55
  56 static void
  57 queue_message_restarted (SoupMessage *msg, gpointer user_data)
  58 {
  59         SoupMessageQueueItem *item = user_data;
  60
  61         g_cancellable_reset (item->cancellable);
  62 }
  63
  64 /**
  65  * soup_message_queue_append:
  66  * @queue: a #SoupMessageQueue
  67  * @msg: a #SoupMessage
  68  * @callback: the callback for @msg
  69  * @user_data: the data to pass to @callback
  70  *
  71  * Creates a new #SoupMessageQueueItem and appends it to @queue.
  72  *
  73  * Return value: the new item, which you must unref with
  74  * soup_message_queue_unref_item() when you are done with.
  75  **/
  76 SoupMessageQueueItem *
  77 soup_message_queue_append (SoupMessageQueue *queue, SoupMessage *msg,
  78                            SoupSessionCallback callback, gpointer user_data)
  79 {
  80         SoupMessageQueueItem *item;
  81
  82         item = g_slice_new0 (SoupMessageQueueItem);
  83         item->session = g_object_ref (queue->session);
  84         item->async_context = soup_session_get_async_context (item->session);
  85         if (item->async_context)
  86                 g_main_context_ref (item->async_context);
  87         item->queue = queue;
  88         item->msg = g_object_ref (msg);
  89         item->callback = callback;
  90         item->callback_data = user_data;
  91         item->cancellable = g_cancellable_new ();
  92         item->priority = soup_message_get_priority (msg);
  93
  94         g_signal_connect (msg, "restarted",
  95                           G_CALLBACK (queue_message_restarted), item);
  96
  97         /* Note: the initial ref_count of 1 represents the caller's
  98          * ref; the queue's own ref is indicated by the absence of the
  99          * "removed" flag.
 100          */
 101         item->ref_count = 1;
 102
 103         g_mutex_lock (&queue->mutex);
 104         if (queue->head) {
 105                 SoupMessageQueueItem *it = queue->head;
 106
 107                 while (it && it->priority >= item->priority)
 108                         it = it->next;
 109
 110                 if (!it) {
 111                         if (queue->tail) {
 112                                 queue->tail->next = item;
 113                                 item->prev = queue->tail;
 114                         } else
 115                                 queue->head = item;
 116                         queue->tail = item;
 117                 } else {
 118                         if (it != queue->head)
 119                                 it->prev->next = item;
 120                         else
 121                                 queue->head = item;
 122                         item->prev = it->prev;
 123                         it->prev = item;
 124                         item->next = it;
 125                 }
 126         } else
 127                 queue->head = queue->tail = item;
 128
 129         g_mutex_unlock (&queue->mutex);
 130         return item;
 131 }
 132
 133 /**
 134  * soup_message_queue_item_ref:
 135  * @item: a #SoupMessageQueueItem
 136  *
 137  * Refs @item.
 138  **/
 139 void
 140 soup_message_queue_item_ref (SoupMessageQueueItem *item)
 141 {
 142         g_mutex_lock (&item->queue->mutex);
 143         item->ref_count++;
 144         g_mutex_unlock (&item->queue->mutex);
 145 }
 146
 147 /**
 148  * soup_message_queue_item_unref:
 149  * @item: a #SoupMessageQueueItem
 150  *
 151  * Unrefs @item; use this on a #SoupMessageQueueItem that you are done
 152  * with (but that you aren't passing to
 153  * soup_message_queue_item_next()).
 154  **/
 155 void
 156 soup_message_queue_item_unref (SoupMessageQueueItem *item)
 157 {
 158         g_mutex_lock (&item->queue->mutex);
 159
 160         /* Decrement the ref_count; if it's still non-zero OR if the
 161          * item is still in the queue, then return.
 162          */
 163         if (--item->ref_count || !item->removed) {
 164                 g_mutex_unlock (&item->queue->mutex);
 165                 return;
 166         }
 167
 168         g_warn_if_fail (item->conn == NULL);
 169
 170         /* OK, @item is dead. Rewrite @queue around it */
 171         if (item->prev)
 172                 item->prev->next = item->next;
 173         else
 174                 item->queue->head = item->next;
 175         if (item->next)
 176                 item->next->prev = item->prev;
 177         else
 178                 item->queue->tail = item->prev;
 179
 180         g_mutex_unlock (&item->queue->mutex);
 181
 182         /* And free it */
 183         g_signal_handlers_disconnect_by_func (item->msg,
 184                                               queue_message_restarted, item);
 185         g_object_unref (item->session);
 186         g_object_unref (item->msg);
 187         g_object_unref (item->cancellable);
 188         g_clear_error (&item->error);
 189         g_clear_object (&item->task);
 190         g_clear_pointer (&item->async_context, g_main_context_unref);
 191         if (item->io_source) {
 192                 g_source_destroy (item->io_source);
 193                 g_source_unref (item->io_source);
 194         }
 195         g_slice_free (SoupMessageQueueItem, item);
 196 }
 197
 198 /**
 199  * soup_message_queue_lookup:
 200  * @queue: a #SoupMessageQueue
 201  * @msg: a #SoupMessage
 202  *
 203  * Finds the #SoupMessageQueueItem for @msg in @queue. You must unref
 204  * the item with soup_message_queue_unref_item() when you are done
 205  * with it.
 206  *
 207  * Return value: (nullable): the queue item for @msg, or %NULL
 208  **/
 209 SoupMessageQueueItem *
 210 soup_message_queue_lookup (SoupMessageQueue *queue, SoupMessage *msg)
 211 {
 212         SoupMessageQueueItem *item;
 213
 214         g_mutex_lock (&queue->mutex);
 215
 216         item = queue->tail;
 217         while (item && (item->removed || item->msg != msg))
 218                 item = item->prev;
 219
 220         if (item)
 221                 item->ref_count++;
 222
 223         g_mutex_unlock (&queue->mutex);
 224         return item;
 225 }
 226
 227 /**
 228  * soup_message_queue_first:
 229  * @queue: a #SoupMessageQueue
 230  *
 231  * Gets the first item in @queue. You must unref the item by calling
 232  * soup_message_queue_unref_item() on it when you are done.
 233  * (soup_message_queue_next() does this for you automatically, so you
 234  * only need to unref the item yourself if you are not going to
 235  * finishing walking the queue.)
 236  *
 237  * Return value: the first item in @queue.
 238  **/
 239 SoupMessageQueueItem *
 240 soup_message_queue_first (SoupMessageQueue *queue)
 241 {
 242         SoupMessageQueueItem *item;
 243
 244         g_mutex_lock (&queue->mutex);
 245
 246         item = queue->head;
 247         while (item && item->removed)
 248                 item = item->next;
 249
 250         if (item)
 251                 item->ref_count++;
 252
 253         g_mutex_unlock (&queue->mutex);
 254         return item;
 255 }
 256
 257 /**
 258  * soup_message_queue_next:
 259  * @queue: a #SoupMessageQueue
 260  * @item: a #SoupMessageQueueItem
 261  *
 262  * Unrefs @item and gets the next item after it in @queue. As with
 263  * soup_message_queue_first(), you must unref the returned item
 264  * yourself with soup_message_queue_unref_item() if you do not finish
 265  * walking the queue.
 266  *
 267  * Return value: the next item in @queue.
 268  **/
 269 SoupMessageQueueItem *
 270 soup_message_queue_next (SoupMessageQueue *queue, SoupMessageQueueItem *item)
 271 {
 272         SoupMessageQueueItem *next;
 273
 274         g_mutex_lock (&queue->mutex);
 275
 276         next = item->next;
 277         while (next && next->removed)
 278                 next = next->next;
 279         if (next)
 280                 next->ref_count++;
 281
 282         g_mutex_unlock (&queue->mutex);
 283         soup_message_queue_item_unref (item);
 284         return next;
 285 }
 286
 287 /**
 288  * soup_message_queue_remove:
 289  * @queue: a #SoupMessageQueue
 290  * @item: a #SoupMessageQueueItem
 291  *
 292  * Removes @item from @queue. Note that you probably also need to call
 293  * soup_message_queue_unref_item() after this.
 294  **/
 295 void
 296 soup_message_queue_remove (SoupMessageQueue *queue, SoupMessageQueueItem *item)
 297 {
 298         g_return_if_fail (!item->removed);
 299
 300         g_mutex_lock (&queue->mutex);
 301         item->removed = TRUE;
 302         g_mutex_unlock (&queue->mutex);
 303 }