// Copyright (C) 2006 Davis E. King (davis@dlib.net)
// License: Boost Software License See LICENSE.txt for the full license.
#ifndef DLIB_PIPE_KERNEl_1_
#define DLIB_PIPE_KERNEl_1_
#include "../algs.h"
#include "../threads.h"
#include "pipe_kernel_abstract.h"
namespace dlib
{
template <
typename T
>
class pipe
{
/*!
INITIAL VALUE
- pipe_size == 0
- pipe_max_size == defined by constructor
- enabled == true
- data == a pointer to an array of ((pipe_max_size>0)?pipe_max_size:1) T objects.
- dequeue_waiters == 0
- enqueue_waiters == 0
- first == 1
- last == 1
- unblock_sig_waiters == 0
CONVENTION
- size() == pipe_size
- max_size() == pipe_max_size
- is_enabled() == enabled
- m == the mutex used to lock access to all the members of this class
- dequeue_waiters == the number of threads blocked on calls to dequeue()
- enqueue_waiters == the number of threads blocked on calls to enqueue() and
wait_until_empty()
- unblock_sig_waiters == the number of threads blocked on calls to
wait_for_num_blocked_dequeues() and the destructor. (i.e. the number of
blocking calls to unblock_sig.wait())
- dequeue_sig == the signaler that threads blocked on calls to dequeue() wait on
- enqueue_sig == the signaler that threads blocked on calls to enqueue()
or wait_until_empty() wait on.
- unblock_sig == the signaler that is signaled when a thread stops blocking on a call
to enqueue() or dequeue(). It is also signaled when a dequeue that will probably
block is called. The destructor and wait_for_num_blocked_dequeues are the only
things that will wait on this signaler.
- if (pipe_size > 0) then
- data[first] == the next item to dequeue
- data[last] == the item most recently added via enqueue, so the last to dequeue.
- else if (pipe_max_size == 0)
- if (first == 0 && last == 0) then
- data[0] == the next item to dequeue
- else if (first == 0 && last == 1) then
- data[0] has been taken out already by a dequeue
!*/
public:
// this is here for backwards compatibility with older versions of dlib.
typedef pipe kernel_1a;
typedef T type;
explicit pipe (
size_t maximum_size
);
virtual ~pipe (
);
void empty (
);
void wait_until_empty (
) const;
void wait_for_num_blocked_dequeues (
unsigned long num
)const;
void enable (
);
void disable (
);
bool is_enqueue_enabled (
) const;
void disable_enqueue (
);
void enable_enqueue (
);
bool is_dequeue_enabled (
) const;
void disable_dequeue (
);
void enable_dequeue (
);
bool is_enabled (
) const;
size_t max_size (
) const;
size_t size (
) const;
bool enqueue (
T& item
);
bool enqueue (
T&& item
) { return enqueue(item); }
bool dequeue (
T& item
);
bool enqueue_or_timeout (
T& item,
unsigned long timeout
);
bool enqueue_or_timeout (
T&& item,
unsigned long timeout
) { return enqueue_or_timeout(item,timeout); }
bool dequeue_or_timeout (
T& item,
unsigned long timeout
);
private:
size_t pipe_size;
const size_t pipe_max_size;
bool enabled;
T* const data;
size_t first;
size_t last;
mutex m;
signaler dequeue_sig;
signaler enqueue_sig;
signaler unblock_sig;
unsigned long dequeue_waiters;
mutable unsigned long enqueue_waiters;
mutable unsigned long unblock_sig_waiters;
bool enqueue_enabled;
bool dequeue_enabled;
// restricted functions
pipe(const pipe&); // copy constructor
pipe& operator=(const pipe&); // assignment operator
};
// ----------------------------------------------------------------------------------------
// ----------------------------------------------------------------------------------------
// member function definitions
// ----------------------------------------------------------------------------------------
// ----------------------------------------------------------------------------------------
template <
typename T
>
pipe<T>::
pipe (
size_t maximum_size
) :
pipe_size(0),
pipe_max_size(maximum_size),
enabled(true),
data(new T[(maximum_size>0) ? maximum_size : 1]),
first(1),
last(1),
dequeue_sig(m),
enqueue_sig(m),
unblock_sig(m),
dequeue_waiters(0),
enqueue_waiters(0),
unblock_sig_waiters(0),
enqueue_enabled(true),
dequeue_enabled(true)
{
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
pipe<T>::
~pipe (
)
{
auto_mutex M(m);
++unblock_sig_waiters;
// first make sure no one is blocked on any calls to enqueue() or dequeue()
enabled = false;
dequeue_sig.broadcast();
enqueue_sig.broadcast();
unblock_sig.broadcast();
// wait for all threads to unblock
while (dequeue_waiters > 0 || enqueue_waiters > 0 || unblock_sig_waiters > 1)
unblock_sig.wait();
delete [] data;
--unblock_sig_waiters;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
empty (
)
{
auto_mutex M(m);
pipe_size = 0;
// let any calls to enqueue() know that the pipe is now empty
if (enqueue_waiters > 0)
enqueue_sig.broadcast();
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
wait_until_empty (
) const
{
auto_mutex M(m);
// this function is sort of like a call to enqueue so treat it like that
++enqueue_waiters;
while (pipe_size > 0 && enabled && dequeue_enabled )
enqueue_sig.wait();
// let the destructor know we are ending if it is blocked waiting
if (enabled == false)
unblock_sig.broadcast();
--enqueue_waiters;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
enable (
)
{
auto_mutex M(m);
enabled = true;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
disable (
)
{
auto_mutex M(m);
enabled = false;
dequeue_sig.broadcast();
enqueue_sig.broadcast();
unblock_sig.broadcast();
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
is_enabled (
) const
{
auto_mutex M(m);
return enabled;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
size_t pipe<T>::
max_size (
) const
{
auto_mutex M(m);
return pipe_max_size;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
size_t pipe<T>::
size (
) const
{
auto_mutex M(m);
return pipe_size;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
enqueue (
T& item
)
{
auto_mutex M(m);
++enqueue_waiters;
// wait until there is room or we are disabled
while (pipe_size == pipe_max_size && enabled && enqueue_enabled &&
!(pipe_max_size == 0 && first == 1) )
enqueue_sig.wait();
if (enabled == false || enqueue_enabled == false)
{
--enqueue_waiters;
// let the destructor know we are unblocking
unblock_sig.broadcast();
return false;
}
// set the appropriate values for first and last
if (pipe_size == 0)
{
first = 0;
last = 0;
}
else
{
last = (last+1)%pipe_max_size;
}
exchange(item,data[last]);
// wake up a call to dequeue() if there are any currently blocked
if (dequeue_waiters > 0)
dequeue_sig.signal();
if (pipe_max_size > 0)
{
++pipe_size;
}
else
{
// wait for a dequeue to take the item out
while (last == 0 && enabled && enqueue_enabled)
enqueue_sig.wait();
if (last == 0 && (enabled == false || enqueue_enabled == false))
{
last = 1;
first = 1;
// no one dequeued this object to put it back into item
exchange(item,data[0]);
--enqueue_waiters;
// let the destructor know we are unblocking
if (unblock_sig_waiters > 0)
unblock_sig.broadcast();
return false;
}
last = 1;
first = 1;
// tell any waiting calls to enqueue() that one of them can proceed
if (enqueue_waiters > 1)
enqueue_sig.broadcast();
// let the destructor know we are unblocking
if (enabled == false && unblock_sig_waiters > 0)
unblock_sig.broadcast();
}
--enqueue_waiters;
return true;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
dequeue (
T& item
)
{
auto_mutex M(m);
++dequeue_waiters;
if (pipe_size == 0)
{
// notify wait_for_num_blocked_dequeues()
if (unblock_sig_waiters > 0)
unblock_sig.broadcast();
// notify any blocked enqueue_or_timeout() calls
if (enqueue_waiters > 0)
enqueue_sig.broadcast();
}
// wait until there is something in the pipe or we are disabled
while (pipe_size == 0 && enabled && dequeue_enabled &&
!(pipe_max_size == 0 && first == 0 && last == 0) )
dequeue_sig.wait();
if (enabled == false || dequeue_enabled == false)
{
--dequeue_waiters;
// let the destructor know we are unblocking
unblock_sig.broadcast();
return false;
}
exchange(item,data[first]);
if (pipe_max_size > 0)
{
// set the appropriate values for first
first = (first+1)%pipe_max_size;
--pipe_size;
}
else
{
// let the enqueue waiting on us know that we took the
// item out already.
last = 1;
}
// wake up a call to enqueue() if there are any currently blocked
if (enqueue_waiters > 0)
enqueue_sig.broadcast();
--dequeue_waiters;
return true;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
enqueue_or_timeout (
T& item,
unsigned long timeout
)
{
auto_mutex M(m);
++enqueue_waiters;
// wait until there is room or we are disabled or
// we run out of time.
bool timed_out = false;
while (pipe_size == pipe_max_size && enabled && enqueue_enabled &&
!(pipe_max_size == 0 && dequeue_waiters > 0 && first == 1) )
{
if (timeout == 0 || enqueue_sig.wait_or_timeout(timeout) == false)
{
timed_out = true;
break;
}
}
if (enabled == false || timed_out || enqueue_enabled == false)
{
--enqueue_waiters;
// let the destructor know we are unblocking
unblock_sig.broadcast();
return false;
}
// set the appropriate values for first and last
if (pipe_size == 0)
{
first = 0;
last = 0;
}
else
{
last = (last+1)%pipe_max_size;
}
exchange(item,data[last]);
// wake up a call to dequeue() if there are any currently blocked
if (dequeue_waiters > 0)
dequeue_sig.signal();
if (pipe_max_size > 0)
{
++pipe_size;
}
else
{
// wait for a dequeue to take the item out
while (last == 0 && enabled && enqueue_enabled)
enqueue_sig.wait();
if (last == 0 && (enabled == false || enqueue_enabled == false))
{
last = 1;
first = 1;
// no one dequeued this object to put it back into item
exchange(item,data[0]);
--enqueue_waiters;
// let the destructor know we are unblocking
if (unblock_sig_waiters > 0)
unblock_sig.broadcast();
return false;
}
last = 1;
first = 1;
// tell any waiting calls to enqueue() that one of them can proceed
if (enqueue_waiters > 1)
enqueue_sig.broadcast();
// let the destructor know we are unblocking
if (enabled == false && unblock_sig_waiters > 0)
unblock_sig.broadcast();
}
--enqueue_waiters;
return true;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
dequeue_or_timeout (
T& item,
unsigned long timeout
)
{
auto_mutex M(m);
++dequeue_waiters;
if (pipe_size == 0)
{
// notify wait_for_num_blocked_dequeues()
if (unblock_sig_waiters > 0)
unblock_sig.broadcast();
// notify any blocked enqueue_or_timeout() calls
if (enqueue_waiters > 0)
enqueue_sig.broadcast();
}
bool timed_out = false;
// wait until there is something in the pipe or we are disabled or we timeout.
while (pipe_size == 0 && enabled && dequeue_enabled &&
!(pipe_max_size == 0 && first == 0 && last == 0) )
{
if (timeout == 0 || dequeue_sig.wait_or_timeout(timeout) == false)
{
timed_out = true;
break;
}
}
if (enabled == false || timed_out || dequeue_enabled == false)
{
--dequeue_waiters;
// let the destructor know we are unblocking
unblock_sig.broadcast();
return false;
}
exchange(item,data[first]);
if (pipe_max_size > 0)
{
// set the appropriate values for first
first = (first+1)%pipe_max_size;
--pipe_size;
}
else
{
// let the enqueue waiting on us know that we took the
// item out already.
last = 1;
}
// wake up a call to enqueue() if there are any currently blocked
if (enqueue_waiters > 0)
enqueue_sig.broadcast();
--dequeue_waiters;
return true;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
wait_for_num_blocked_dequeues (
unsigned long num
)const
{
auto_mutex M(m);
++unblock_sig_waiters;
while ( (dequeue_waiters < num || pipe_size != 0) && enabled && dequeue_enabled)
unblock_sig.wait();
// let the destructor know we are ending if it is blocked waiting
if (enabled == false)
unblock_sig.broadcast();
--unblock_sig_waiters;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
is_enqueue_enabled (
) const
{
auto_mutex M(m);
return enqueue_enabled;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
disable_enqueue (
)
{
auto_mutex M(m);
enqueue_enabled = false;
enqueue_sig.broadcast();
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
enable_enqueue (
)
{
auto_mutex M(m);
enqueue_enabled = true;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
bool pipe<T>::
is_dequeue_enabled (
) const
{
auto_mutex M(m);
return dequeue_enabled;
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
disable_dequeue (
)
{
auto_mutex M(m);
dequeue_enabled = false;
dequeue_sig.broadcast();
}
// ----------------------------------------------------------------------------------------
template <
typename T
>
void pipe<T>::
enable_dequeue (
)
{
auto_mutex M(m);
dequeue_enabled = true;
}
// ----------------------------------------------------------------------------------------
}
#endif // DLIB_PIPE_KERNEl_1_