src/ptclib/qchannel.cxx

   1 /*
   2  * qchannel.cxx
   3  *
   4  * Class for implementing a serial queue channel in memory.
   5  *
   6  * Portable Windows Library
   7  *
   8  * Copyright (c) 2001 Equivalence Pty. Ltd.
   9  *
  10  * The contents of this file are subject to the Mozilla Public License
  11  * Version 1.0 (the "License"); you may not use this file except in
  12  * compliance with the License. You may obtain a copy of the License at
  13  * http://www.mozilla.org/MPL/
  14  *
  15  * Software distributed under the License is distributed on an "AS IS"
  16  * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
  17  * the License for the specific language governing rights and limitations
  18  * under the License.
  19  *
  20  * The Original Code is Portable Windows Library.
  21  *
  22  * The Initial Developer of the Original Code is Equivalence Pty. Ltd.
  23  *
  24  * Contributor(s): ______________________________________.
  25  *
  26  * $Log$
  27  * Revision 1.2  2001/09/10 02:51:23  robertj
  28  * Major change to fix problem with error codes being corrupted in a
  29  *   PChannel when have simultaneous reads and writes in threads.
  30  *
  31  * Revision 1.1  2001/07/10 03:07:07  robertj
  32  * Added queue channel and delay channel classes to ptclib.
  33  *
  34  */
  35
  36 #ifdef __GNUC__
  37 #pragma implementation "qchannel.h"
  38 #endif
  39
  40 #include <ptlib.h>
  41 #include <ptclib/qchannel.h>
  42
  43
  44 #define new PNEW
  45
  46
  47 /////////////////////////////////////////////////////////
  48
  49 PQueueChannel::PQueueChannel(PINDEX size)
  50 {
  51   if (size > 0) {
  52     queueBuffer = new BYTE[size];
  53     os_handle = 1;
  54   }
  55   else {
  56     queueBuffer = NULL;
  57     os_handle = -1;
  58   }
  59   queueSize = size;
  60   queueLength = enqueuePos = dequeuePos = 0;
  61 }
  62
  63
  64 PQueueChannel::~PQueueChannel()
  65 {
  66   delete queueBuffer;
  67 }
  68
  69
  70 BOOL PQueueChannel::Open(PINDEX size)
  71 {
  72   if (size == 0)
  73     Close();
  74   else {
  75     mutex.Wait();
  76     delete queueBuffer;
  77     queueBuffer = new BYTE[size];
  78     queueSize = size;
  79     queueLength = enqueuePos = dequeuePos = 0;
  80     os_handle = 1;
  81     mutex.Signal();
  82
  83     unempty.Signal();
  84     unfull.Signal();
  85   }
  86
  87   return TRUE;
  88 }
  89
  90
  91 BOOL PQueueChannel::Close()
  92 {
  93   if (!IsOpen())
  94     return FALSE;
  95
  96   mutex.Wait();
  97   delete queueBuffer;
  98   queueBuffer = NULL;
  99   os_handle = -1;
 100   mutex.Signal();
 101   unempty.Signal();
 102   unfull.Signal();
 103   return TRUE;
 104 }
 105
 106
 107 BOOL PQueueChannel::Read(void * buf, PINDEX count)
 108 {
 109   lastReadCount = 0;
 110
 111   if (!IsOpen())
 112     return FALSE;
 113
 114   BYTE * buffer = (BYTE *)buf;
 115
 116   while (count > 0) {
 117
 118     mutex.Wait();
 119
 120     /* If queue is empty then we should block for the time specifed in the
 121        read timeout.
 122      */
 123     while (queueLength == 0) {
 124       mutex.Signal();
 125
 126       PTRACE_IF(6, readTimeout > 0, "QChan\tBlocking on empty queue");
 127       if (!unempty.Wait(readTimeout)) {
 128         PTRACE(6, "QChan\tRead timeout on empty queue");
 129         if (lastReadCount == 0)
 130           return SetErrorValues(Timeout, EAGAIN, LastReadError);
 131         return TRUE;
 132       }
 133
 134       if (!IsOpen())
 135         return SetErrorValues(Interrupted, EINTR, LastReadError);
 136
 137       mutex.Wait();
 138     }
 139
 140     // To make things simpler, limit to amount to copy out of queue to till
 141     // the end of the linear part of memory. Another loop around will get
 142     // rest of data to dequeue
 143     PINDEX copyLen = queueSize - dequeuePos;
 144
 145     // But do not copy more than has actually been queued
 146     if (copyLen > queueLength)
 147       copyLen = queueLength;
 148
 149     // Or more than has been requested
 150     if (copyLen > count)
 151       copyLen = count;
 152
 153     // Copy data out and increment pointer, decrement bytes yet to dequeue
 154     memcpy(buffer, queueBuffer+dequeuePos, copyLen);
 155     lastReadCount += copyLen;
 156     buffer += copyLen;
 157     count -= copyLen;
 158
 159     // Move the queue pointer along, wrapping to beginning
 160     dequeuePos += copyLen;
 161     if (dequeuePos >= queueSize)
 162       dequeuePos = 0;
 163
 164     // If buffer was full, signal possibly blocked write of data to queue
 165     // that it can write to queue now.
 166     if (queueLength == queueSize) {
 167       PTRACE(6, "QChan\tSignalling queue no longer full");
 168       unfull.Signal();
 169     }
 170
 171     // Now decrement queue length by the amount we copied
 172     queueLength -= copyLen;
 173
 174     mutex.Signal();
 175
 176   }
 177
 178   return TRUE;
 179 }
 180
 181
 182 BOOL PQueueChannel::Write(const void * buf, PINDEX count)
 183 {
 184   lastWriteCount = 0;
 185
 186   if (!IsOpen())
 187     return FALSE;
 188
 189   const BYTE * buffer = (BYTE *)buf;
 190
 191   while (count > 0) {
 192
 193     mutex.Wait();
 194
 195     /* If queue is full then we should block for the time specifed in the
 196        write timeout.
 197      */
 198     while (queueLength == queueSize) {
 199       mutex.Signal();
 200
 201       PTRACE_IF(6, writeTimeout > 0, "QChan\tBlocking on full queue");
 202       if (!unfull.Wait(writeTimeout)) {
 203         PTRACE(6, "QChan\tWrite timeout on full queue");
 204         return SetErrorValues(Timeout, EAGAIN, LastWriteError);
 205       }
 206
 207       if (!IsOpen())
 208         return SetErrorValues(Interrupted, EINTR, LastWriteError);
 209
 210       mutex.Wait();
 211     }
 212
 213     // Calculate number of bytes to copy
 214     PINDEX copyLen = count;
 215
 216     // First don't copy more than are availble in queue
 217     PINDEX bytesLeftInQueue = queueSize - queueLength;
 218     if (copyLen > bytesLeftInQueue)
 219       copyLen = bytesLeftInQueue;
 220
 221     // Then to make things simpler, limit to amount left till the end of the
 222     // linear part of memory. Another loop around will get rest of data to queue
 223     PINDEX bytesLeftInUnwrapped = queueSize - enqueuePos;
 224     if (copyLen > bytesLeftInUnwrapped)
 225       copyLen = bytesLeftInUnwrapped;
 226
 227     // Move the data in and increment pointer, decrement bytes yet to queue
 228     memcpy(queueBuffer + enqueuePos, buffer, copyLen);
 229     lastWriteCount += copyLen;
 230     buffer += copyLen;
 231     count -= copyLen;
 232
 233     // Move the queue pointer along, wrapping to beginning
 234     enqueuePos += copyLen;
 235     if (enqueuePos >= queueSize)
 236       enqueuePos = 0;
 237
 238     // If buffer was empty, signal possibly blocked reader of data from queue
 239     // that it can read from queue now.
 240     if (queueLength == 0) {
 241       PTRACE(6, "QChan\tSignalling queue no longer empty");
 242       unempty.Signal();
 243     }
 244
 245     // Now increment queue length by the amount we copied
 246     queueLength += copyLen;
 247
 248     mutex.Signal();
 249   }
 250
 251   return TRUE;
 252 }
 253
 254
 255 // End of File ///////////////////////////////////////////////////////////////