summaryrefslogtreecommitdiff
path: root/ace/Message_Queue.h
blob: 07c795f20b16594c453d98fabd797d0363313cba (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
// -*- C++ -*-

//=============================================================================
/**
 *  @file    Message_Queue.h
 *
 *  $Id$
 *
 *  @author Douglas C. Schmidt <schmidt@cs.wustl.edu>
 */
//=============================================================================

#ifndef ACE_MESSAGE_QUEUE_H
#define ACE_MESSAGE_QUEUE_H
#include /**/ "ace/pre.h"

#include "ace/Message_Block.h"

#if !defined (ACE_LACKS_PRAGMA_ONCE)
# pragma once
#endif /* ACE_LACKS_PRAGMA_ONCE */

#include "ace/IO_Cntl_Msg.h"
#include "ace/Synch.h"

// Forward decls.
class ACE_Notification_Strategy;
template <ACE_SYNCH_DECL> class ACE_Message_Queue_Iterator;
template <ACE_SYNCH_DECL> class ACE_Message_Queue_Reverse_Iterator;

/**
 * @class ACE_Message_Queue_Base
 *
 * @brief Base class for <ACE_Message_Queue>, which is the central
 * queueing facility for messages in the ACE framework.
 *
 * For all the <ACE_Time_Value> pointer parameters the caller will
 * block until action is possible if <timeout> == 0.  Otherwise, it
 * will wait until the absolute time specified in *<timeout>
 * elapses.
 *
 * A queue is always in one of three states:
 * . ACTIVATED
 * . DEACTIVATED
 * . PULSED
 */
class ACE_Export ACE_Message_Queue_Base
{
public:
  enum
  {
    // Default high and low watermarks.

    /// Default high watermark (16 K).
    DEFAULT_HWM = 16 * 1024,
    /// Default low watermark (same as high water mark).
    DEFAULT_LWM = 16 * 1024,

    // Queue states.  Before PULSED state was added, the activate()
    // and deactivate() methods returned WAS_INACTIVE or WAS_ACTIVE
    // to indicate the previous condition.  Now those methods
    // return the state the queue was previously in.  WAS_ACTIVE
    // and WAS_INACTIVE are defined to match previous semantics for
    // applications that don't use the PULSED state.

    WAS_ACTIVE = 1, /* DEPRECATED */
    /// Message queue is active and processing normally
    ACTIVATED = 1,

    WAS_INACTIVE = 2, /* DEPRECATED */
    /// Queue is deactivated; no enqueue or dequeue operations allowed.
    DEACTIVATED = 2,

    /// Message queue was pulsed; enqueue and dequeue may proceed normally.
    PULSED = 3

  };

  ACE_Message_Queue_Base (void);

  /// Close down the message queue and release all resources.
  virtual int close (void) = 0;

  /// Close down the message queue and release all resources.
  virtual ~ACE_Message_Queue_Base (void);

  // = Enqueue and dequeue methods.

  /**
   * Retrieve the first <ACE_Message_Block> without removing it.  Note
   * that <timeout> uses <{absolute}> time rather than <{relative}>
   * time.  If the <timeout> elapses without receiving a message -1 is
   * returned and <errno> is set to <EWOULDBLOCK>.  If the queue is
   * deactivated -1 is returned and <errno> is set to <ESHUTDOWN>.
   * Otherwise, returns -1 on failure, else the number of items still
   * on the queue.
   */
  virtual int peek_dequeue_head (ACE_Message_Block *&first_item,
                                 ACE_Time_Value *timeout = 0) = 0;

  /**
   * Enqueue a <ACE_Message_Block *> into the tail of the queue.
   * Returns number of items in queue if the call succeeds or -1
   * otherwise.  These calls return -1 when queue is closed,
   * deactivated (in which case <errno> == <ESHUTDOWN>), when a signal
   * occurs (in which case <errno> == <EINTR>, or if the time
   * specified in timeout elapses (in which case <errno> ==
   * <EWOULDBLOCK>).
   */
  virtual int enqueue_tail (ACE_Message_Block *new_item,
                            ACE_Time_Value *timeout = 0) = 0;
  virtual int enqueue (ACE_Message_Block *new_item,
                       ACE_Time_Value *timeout = 0) = 0;

  /**
   * Dequeue and return the <ACE_Message_Block *> at the head of the
   * queue.  Returns number of items in queue if the call succeeds or
   * -1 otherwise.  These calls return -1 when queue is closed,
   * deactivated (in which case <errno> == <ESHUTDOWN>), when a signal
   * occurs (in which case <errno> == <EINTR>, or if the time
   * specified in timeout elapses (in which case <errno> ==
   * <EWOULDBLOCK>).
   */
  virtual int dequeue_head (ACE_Message_Block *&first_item,
                            ACE_Time_Value *timeout = 0) = 0;
  virtual int dequeue (ACE_Message_Block *&first_item,
                       ACE_Time_Value *timeout = 0) = 0;

  // = Check if queue is full/empty.
  /// True if queue is full, else false.
  virtual int is_full (void) = 0;

  /// True if queue is empty, else false.
  virtual int is_empty (void) = 0;

  // = Queue statistic methods.

  /// Number of total bytes on the queue, i.e., sum of the message
  /// block sizes.
  virtual size_t message_bytes (void) = 0;

  /// Number of total length on the queue, i.e., sum of the message
  /// block lengths.
  virtual size_t message_length (void) = 0;

  /// Number of total messages on the queue.
  virtual int message_count (void) = 0;

  /// New value of the number of total bytes on the queue, i.e.,
  /// sum of the message block sizes.
  virtual void message_bytes (size_t new_size) = 0;

  /// New value of the number of total length on the queue, i.e.,
  /// sum of the message block lengths.
  virtual void message_length (size_t new_length) = 0;

  // = Activation control methods.

  /**
   * Deactivate the queue and wake up all threads waiting on the queue
   * so they can continue.  No messages are removed from the queue,
   * however.  Any other operations called until the queue is
   * activated again will immediately return -1 with @c errno
   * ESHUTDOWN.
   *
   * @retval  The queue's state before this call.
   */
  virtual int deactivate (void) = 0;

  /**
   * Reactivate the queue so that threads can enqueue and dequeue
   * messages again.
   *
   * @retval  The queue's state before this call.
   */
  virtual int activate (void) = 0;

  /**
   * Pulse the queue to wake up any waiting threads.  Changes the
   * queue state to PULSED; future enqueue/dequeue operations proceed
   * as in ACTIVATED state.
   *
   * @retval  The queue's state before this call.
   */
  virtual int pulse (void) = 0;

  /// Returns the current state of the queue.
  virtual int state (void);

  /// Returns 1 if the state of the queue is DEACTIVATED,
  /// and 0 if the queue's state is ACTIVATED or PULSED.
  virtual int deactivated (void) = 0;

  /// Get the notification strategy for the <Message_Queue>
  virtual ACE_Notification_Strategy *notification_strategy (void) = 0;

  /// Set the notification strategy for the <Message_Queue>
  virtual void notification_strategy (ACE_Notification_Strategy *s) = 0;

  // = Notification hook.

  /// Dump the state of an object.
  virtual void dump (void) const = 0;

  /// Declare the dynamic allocation hooks.
  ACE_ALLOC_HOOK_DECLARE;

protected:
  /// Indicates the state of the queue, which can be
  /// <ACTIVATED>, <DEACTIVATED>, or <PULSED>.
  int state_;

private:
  // = Disallow these operations.
  ACE_UNIMPLEMENTED_FUNC (void operator= (const ACE_Message_Queue_Base &))
  ACE_UNIMPLEMENTED_FUNC (ACE_Message_Queue_Base (const ACE_Message_Queue_Base &))
};

// Include the templates here.
#include "ace/Message_Queue_T.h"

#if defined (VXWORKS)
# include /**/ <msgQLib.h>

/**
 * @class ACE_Message_Queue_Vx
 *
 * @brief Wrapper for VxWorks message queues.
 *
 * Specialization of ACE_Message_Queue to simply wrap VxWorks
 * MsgQ.  It does not use any synchronization, because it relies
 * on the native MsgQ implementation to take care of that.  The
 * only system calls that it uses are VxWorks msgQLib calls, so
 * it is suitable for use in interrupt service routines.
 * NOTE: *Many* ACE_Message_Queue features are not supported with
 * this specialization, including:
 * * The two size arguments to the constructor and <open> are
 * interpreted differently.  The first is interpreted as the
 * maximum number of bytes in a message.  The second is
 * interpreted as the maximum number of messages that can be
 * queued.
 * * <dequeue_head> *requires* that the ACE_Message_Block
 * pointer argument point to an ACE_Message_Block that was
 * allocated by the caller.  It must be big enough to support
 * the received message, without using continuation. The
 * pointer argument is not modified.
 * * Message priority.  MSG_Q_FIFO is hard-coded.
 * * enqueue method timeouts.
 * * <peek_dequeue_head>.
 * * <ACE_Message_Queue_Iterators>.
 * * The ability to change low and high water marks after creation.
 * * <Message_Block> chains.  The continuation field of <ACE_Message_Block>
 * *   is ignored; only the first block of a fragment chain is
 * *   recognized.
 */
class ACE_Message_Queue_Vx : public ACE_Message_Queue<ACE_NULL_SYNCH>
{
public:
  // = Initialization and termination methods.
  ACE_Message_Queue_Vx (size_t max_messages,
                        size_t max_message_length,
                        ACE_Notification_Strategy * = 0);

  // Create a message queue with all the defaults.
  /// Create a message queue with all the defaults.
  virtual int open (size_t max_messages,
                    size_t max_message_length,
                    ACE_Notification_Strategy * = 0);

  /// Close down the message queue and release all resources.
  virtual int close (void);

  /// Close down the message queue and release all resources.
  virtual ~ACE_Message_Queue_Vx (void);

  // = Queue statistic methods.
  /**
   * Number of total bytes on the queue, i.e., sum of the message
   * block sizes.
   */
  virtual size_t message_bytes (void);

  /**
   * Number of total length on the queue, i.e., sum of the message
   * block lengths.
   */
  virtual size_t message_length (void);

  /**
   * Number of total messages on the queue.
   */
  virtual int message_count (void);

  // = Manual changes to these stats (used when queued message blocks
  // change size or lengths).
  /**
   * New value of the number of total bytes on the queue, i.e., sum of
   * the message block sizes.
   */
  virtual void message_bytes (size_t new_size);
  /**
   * New value of the number of total length on the queue, i.e., sum
   * of the message block lengths.
   */
  virtual void message_length (size_t new_length);

  // = Flow control routines

  /// Get high watermark.
  virtual size_t high_water_mark (void);

  /// Set high watermark.
  virtual void high_water_mark (size_t hwm);

  /// Get low watermark.
  virtual size_t low_water_mark (void);

  /// Set low watermark.
  virtual void low_water_mark (size_t lwm);

  // = Activation control methods.

  /// Dump the state of an object.
  void dump (void) const;

  /// Declare the dynamic allocation hooks.
  ACE_ALLOC_HOOK_DECLARE;

protected:
  /// Enqueue an <ACE_Message_Block *> in accordance with its priority.
  virtual int enqueue_i (ACE_Message_Block *new_item);

  /// Enqueue an <ACE_Message_Block *> in accordance with its deadline time.
  virtual int enqueue_deadline_i (ACE_Message_Block *new_item);

  /// Enqueue an <ACE_Message_Block *> at the end of the queue.
  virtual int enqueue_tail_i (ACE_Message_Block *new_item);

  /// Enqueue an <ACE_Message_Block *> at the head of the queue.
  virtual int enqueue_head_i (ACE_Message_Block *new_item);

  /// Dequeue and return the <ACE_Message_Block *> at the head of the
  /// queue.
  virtual int dequeue_head_i (ACE_Message_Block *&first_item);

  /// Dequeue and return the <ACE_Message_Block *> with the lowest
  /// priority.
  virtual int dequeue_prio_i (ACE_Message_Block *&dequeued);

  /// Dequeue and return the <ACE_Message_Block *> at the tail of the
  /// queue.
  virtual int dequeue_tail_i (ACE_Message_Block *&dequeued);

  /// Dequeue and return the <ACE_Message_Block *> that has the lowest
  /// deadline time.
  virtual int dequeue_deadline_i (ACE_Message_Block *&dequeued);

  // = Check the boundary conditions (assumes locks are held).
  /// True if queue is full, else false.
  virtual int is_full_i (void);

  /// True if queue is empty, else false.
  virtual int is_empty_i (void);

  // = Implementation of public <activate>/<deactivate> methods above.

  // These methods assume locks are held.

  // = Helper methods to factor out common #ifdef code.
  /// Wait for the queue to become non-full.
  virtual int wait_not_full_cond (ACE_Guard<ACE_Null_Mutex> &mon,
                                  ACE_Time_Value *tv);

  /// Wait for the queue to become non-empty.
  virtual int wait_not_empty_cond (ACE_Guard<ACE_Null_Mutex> &mon,
                                   ACE_Time_Value *tv);

  /// Inform any threads waiting to enqueue that they can procede.
  virtual int signal_enqueue_waiters (void);

  /// Inform any threads waiting to dequeue that they can procede.
  virtual int signal_dequeue_waiters (void);

  /// Access the underlying msgQ.
  MSG_Q_ID msgq (void);

private:
  /// Maximum number of messages that can be queued.
  int max_messages_;

  /// Maximum message size, in bytes.
  int max_message_length_;

  /// Native message queue options.
  int options_;

  // = Disallow these operations.
  ACE_UNIMPLEMENTED_FUNC (void operator= (const ACE_Message_Queue_Vx &))
  ACE_UNIMPLEMENTED_FUNC (ACE_Message_Queue_Vx (const ACE_Message_Queue_Vx &))

  ACE_UNIMPLEMENTED_FUNC (virtual int peek_dequeue_head
                            (ACE_Message_Block *&first_item,
                             ACE_Time_Value *tv = 0))
};
#endif /* VXWORKS */

#if defined (ACE_WIN32) && (ACE_HAS_WINNT4 != 0)
/**
 * @class ACE_Message_Queue_NT
 *
 * @brief Message Queue implementation using IO completion port on NT.
 *
 * Implementation of a strip-downed ACE_Message_Queue using NT's
 * IO completion port mechanism.
 * NOTE: *Many* ACE_Message_Queue features are not supported with
 * this implementation, including:
 * * <open> method have different signatures.
 * * <dequeue_head> *requires* that the <ACE_Message_Block>
 * pointer argument point to an <ACE_Message_Block> that was
 * allocated by the caller.
 * * <peek_dequeue_head>.
 * * <ACE_Message_Queue_Iterators>.
 * * No flow control.
 */
class ACE_Export ACE_Message_Queue_NT : public ACE_Message_Queue_Base
{
public:
  // = Initialization and termination methods.
  ACE_Message_Queue_NT (DWORD max_threads = ACE_Message_Queue_Base::DEFAULT_HWM);

  /**
   * Initialize the Message Queue by creating a new NT I/O completion
   * port.  The first arguemnt specifies the number of threads
   * released by the MQ that are allowed to run concurrently.  Return
   * 0 when succeeds, -1 otherwise.
   */
  virtual int open (DWORD max_threads = ACE_Message_Queue_Base::DEFAULT_HWM);

  /// Close down the underlying I/O completion port.  You need to
  /// re-open the MQ after this function is executed.
  virtual int close (void);

  /// Close down the message queue and release all resources.
  virtual ~ACE_Message_Queue_NT (void);

  // = Enqueue and dequeue methods.

  /**
   * Enqueue an <ACE_Message_Block *> at the end of the queue.
   * Returns -1 on failure, else the number of items still on the
   * queue.
   */
  virtual int enqueue_tail (ACE_Message_Block *new_item,
                            ACE_Time_Value *timeout = 0);
  virtual int enqueue (ACE_Message_Block *new_item,
                       ACE_Time_Value *timeout = 0);

  /**
   * Dequeue and return the <ACE_Message_Block *> at the head of the
   * queue.  Returns -1 on failure, else the number of items still on
   * the queue.
   */
  virtual int dequeue_head (ACE_Message_Block *&first_item,
                            ACE_Time_Value *timeout = 0);
  virtual int dequeue (ACE_Message_Block *&first_item,
                       ACE_Time_Value *timeout = 0);

  // = Check if queue is full/empty.
  /**
   * Always return false.
   */

  virtual int is_full (void);
  /**
   * True if queue is empty, else false.  Notice the return value is
   * only transient.
   */
  virtual int is_empty (void);

  // = Queue statistic methods (transient.)
  /**
   * Number of total bytes on the queue, i.e., sum of the message
   * block sizes.
   */
  virtual size_t message_bytes (void);

  /**
   * Number of total length on the queue, i.e., sum of the message
   * block lengths.
   */
  virtual size_t message_length (void);

  /**
   * Number of total messages on the queue.
   */
  virtual int message_count (void);

  // = Manual changes to these stats (used when queued message blocks
  // change size or lengths).
  /**
   * New value of the number of total bytes on the queue, i.e., sum of
   * the message block sizes.
   */
  virtual void message_bytes (size_t new_size);

  /**
   * New value of the number of total length on the queue, i.e., sum
   * of the message block lengths.
   */
  virtual void message_length (size_t new_length);

  /// Get the max concurrent thread number.
  virtual DWORD max_threads (void);

  // = Activation control methods.

  /**
   * Deactivate the queue and wake up all threads waiting on the queue
   * so they can continue.  No messages are removed from the queue,
   * however.  Any other operations called until the queue is
   * activated again will immediately return -1 with @c errno
   * ESHUTDOWN.
   *
   * @retval  The queue's state before this call.
   */
  virtual int deactivate (void);

  /**
   * Reactivate the queue so that threads can enqueue and dequeue
   * messages again.  Returns the state of the queue before the call.
   */
  virtual int activate (void);

  /**
   * Pulse the queue to wake up any waiting threads.  Changes the
   * queue state to PULSED; future enqueue/dequeue operations proceed
   * as in ACTIVATED state.
   *
   * @retval  The queue's state before this call.
   */
  virtual int pulse (void);

  /// Returns true if the state of the queue is <DEACTIVATED>,
  /// but false if the queue's is <ACTIVATED> or <PULSED>.
  virtual int deactivated (void);

  // = Not currently implemented...
  int peek_dequeue_head (ACE_Message_Block *&first_item,
                         ACE_Time_Value *timeout = 0);
  ACE_Notification_Strategy *notification_strategy (void);
  void notification_strategy (ACE_Notification_Strategy *s);

  // = Notification hook.

  /// Dump the state of an object.
  virtual void dump (void) const;

  /// Get the handle to the underlying completion port.
  virtual ACE_HANDLE completion_port (void);

  /// Declare the dynamic allocation hooks.
  ACE_ALLOC_HOOK_DECLARE;

private:
  // = Internal states.

  /// Maximum threads that can be released (and run) concurrently.
  DWORD max_cthrs_;

  /// Current number of threads waiting to dequeue messages.
  DWORD cur_thrs_;

  /// Current number of bytes in queue.
  size_t cur_bytes_;

  /// Current length of messages in queue.
  size_t cur_length_;

  /// Current number of messages in the queue.
  int cur_count_;

  /**
   * Synchronizer.  This should really be an ACE_Recursive_Thread_Mutex
   * but since this class is only supported on NT, it's okay to use
   * ACE_Thread_Mutex here.
   */
  ACE_Thread_Mutex lock_;

  /// Underlying NT IoCompletionPort.
  ACE_HANDLE completion_port_;

  // = Disallow these operations.
  ACE_UNIMPLEMENTED_FUNC (void operator= (const ACE_Message_Queue_NT &))
  ACE_UNIMPLEMENTED_FUNC (ACE_Message_Queue_NT (const ACE_Message_Queue_NT &))
};
#endif /* ACE_WIN32 && ACE_HAS_WINNT4 != 0 */


#if defined (__ACE_INLINE__)
#include "ace/Message_Queue.i"
#endif /* __ACE_INLINE__ */

#include /**/ "ace/post.h"
#endif /* ACE_MESSAGE_QUEUE_H */