2 * mono-cq.c: concurrent queue
5 * Gonzalo Paniagua Javier (gonzalo@novell.com)
7 * Copyright (c) 2011 Novell, Inc (http://www.novell.com)
10 #include <mono/metadata/object.h>
11 #include <mono/metadata/mono-cq.h>
12 #include <mono/metadata/mono-mlist.h>
15 //#define CQ_DEBUG(...) g_message(__VA_ARGS__)
20 volatile gint32 count;
23 /* matches the System.MonoListItem object */
30 /* matches the System.MonoCQItem object */
33 MonoArray *array; // MonoObjects
34 MonoArray *array_state; // byte array
35 volatile gint32 first;
39 typedef struct _MonoCQItem MonoCQItem;
40 #define CQ_ARRAY_SIZE 64
42 static MonoVTable *monocq_item_vtable = NULL;
45 mono_cqitem_alloc (void)
48 MonoDomain *domain = mono_get_root_domain ();
50 if (!monocq_item_vtable) {
51 MonoClass *klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoCQItem");
52 monocq_item_vtable = mono_class_vtable (domain, klass);
53 g_assert (monocq_item_vtable);
55 queue = (MonoCQItem *) mono_object_new_fast (monocq_item_vtable);
56 MONO_OBJECT_SETREF (queue, array, mono_array_new (domain, mono_defaults.object_class, CQ_ARRAY_SIZE));
57 MONO_OBJECT_SETREF (queue, array_state, mono_array_new (domain, mono_defaults.byte_class, CQ_ARRAY_SIZE));
66 cq = g_new0 (MonoCQ, 1);
67 MONO_GC_REGISTER_ROOT (cq->head);
68 MONO_GC_REGISTER_ROOT (cq->tail);
69 cq->head = mono_mlist_alloc ((MonoObject *) mono_cqitem_alloc ());
71 CQ_DEBUG ("Created %p", cq);
76 mono_cq_destroy (MonoCQ *cq)
78 CQ_DEBUG ("Destroy %p", cq);
82 mono_gc_bzero (cq, sizeof (MonoCQ));
83 MONO_GC_UNREGISTER_ROOT (cq->tail);
84 MONO_GC_UNREGISTER_ROOT (cq->head);
89 mono_cq_count (MonoCQ *cq)
94 CQ_DEBUG ("Count %d", cq->count);
99 mono_cq_add_node (MonoCQ *cq)
102 MonoMList *prev_tail;
104 CQ_DEBUG ("Adding node");
105 n = mono_mlist_alloc ((MonoObject *) mono_cqitem_alloc ());
106 prev_tail = cq->tail;
107 MONO_OBJECT_SETREF (prev_tail, next, n);
112 mono_cqitem_try_enqueue (MonoCQ *cq, MonoObject *obj)
119 queue = (MonoCQItem *) tail->data;
122 if (pos >= CQ_ARRAY_SIZE) {
123 CQ_DEBUG ("enqueue(): pos >= CQ_ARRAY_SIZE, %d >= %d", pos, CQ_ARRAY_SIZE);
127 if (InterlockedCompareExchange (&queue->last, pos + 1, pos) == pos) {
128 mono_array_setref (queue->array, pos, obj);
129 mono_array_set (queue->array_state, char, pos, TRUE);
130 if ((pos + 1) == CQ_ARRAY_SIZE) {
131 CQ_DEBUG ("enqueue(): pos + 1 == CQ_ARRAY_SIZE, %d. Adding node.", CQ_ARRAY_SIZE);
132 mono_cq_add_node (cq);
137 g_assert_not_reached ();
141 mono_cq_enqueue (MonoCQ *cq, MonoObject *obj)
143 if (cq == NULL || obj == NULL)
147 if (mono_cqitem_try_enqueue (cq, obj)) {
148 CQ_DEBUG ("Queued one");
149 InterlockedIncrement (&cq->count);
157 mono_cq_remove_node (MonoCQ *cq)
161 CQ_DEBUG ("Removing node");
163 /* Not needed now that array_state is GC memory
167 queue = (MonoCQItem *) old_head->data;
170 for (i = 0; i < CQ_ARRAY_SIZE; i++) {
171 if (mono_array_get (queue->array_state, char, i) == TRUE) {
180 while (old_head->next == NULL)
182 cq->head = old_head->next;
187 mono_cqitem_try_dequeue (MonoCQ *cq, MonoObject **obj)
194 queue = (MonoCQItem *) head->data;
197 if (pos >= queue->last || pos >= CQ_ARRAY_SIZE)
200 if (InterlockedCompareExchange (&queue->first, pos + 1, pos) == pos) {
201 while (mono_array_get (queue->array_state, char, pos) == FALSE) {
204 *obj = mono_array_get (queue->array, MonoObject *, pos);
205 mono_array_set (queue->array, MonoObject *, pos, NULL);
206 mono_array_set (queue->array_state, char, pos, FALSE);
207 if ((pos + 1) == CQ_ARRAY_SIZE) {
208 mono_cq_remove_node (cq);
213 g_assert_not_reached ();
217 mono_cq_dequeue (MonoCQ *cq, MonoObject **result)
219 while (cq->count > 0) {
220 if (mono_cqitem_try_dequeue (cq, result)) {
221 CQ_DEBUG ("Dequeued one");
222 InterlockedDecrement (&cq->count);