كيفية تنفيذ حجب قراءة باستخدام POSIX المواضيع

Question

أود أن تنفيذ المنتج/المستهلك السيناريو الذي يطيع الواجهات التي هي تقريبا:

class Consumer {
private:
    vector<char> read(size_t n) {
        // If the internal buffer has `n` elements, then dequeue them
        // Otherwise wait for more data and try again
    }
public:
    void run() {
        read(10);
        read(4839);
        // etc
    }
    void feed(const vector<char> &more) {
        // Safely queue the data
        // Notify `read` that there is now more data
    }
};

في هذه الحالة ، feed و run سوف تعمل على فصل المواضيع read يجب أن يكون حجب قراءة (مثل recv و fread).من الواضح أنني سوف تحتاج إلى نوع من الإقصاء المتبادل على deque و سوف تحتاج الى بعض النوع من إعلام النظام إلى إعلام read في محاولة مرة أخرى.

سمعت حالة المتغيرات هي وسيلة للذهاب ، ولكن كل ما عندي خاصية تعدد تجربة يكمن مع ويندوز أواجه صعوبة في فهم لهم.

شكرا على أي مساعدة!

(نعم, أعلم أنها غير فعالة في العودة ناقلات.دعونا لا ندخل في ذلك.)

Answer 1

هذا الرمز ليس من إنتاج جاهزة.لا التحقق من الخطأ القيام به على النتائج من أي مكتبة المكالمات.

علي لف قفل/فتح من مزامنة في LockThread لذلك هو استثناء آمنة.ولكن ذلك حول هذا الموضوع.

وبالإضافة إلى ذلك إذا كنت تفعل هذا على محمل الجد أود أن التفاف مزامنة والاحكام المتغيرات داخل الكائنات حتى يتمكنوا من المهد يساء استخدامها داخل أساليب أخرى من المستهلكين.ولكن طالما كنت تأخذ علما بأن القفل يجب الحصول عليها قبل استخدام حالة متغير (في أي اتجاه) ثم بسيطة هذه الحالة يمكن أن يقف كما هو.

من الفائدة هل تحققت دفعة خيوط المكتبة ؟

#include <iostream>
#include <vector>
#include <pthread.h>

class LockThread
{
    public:
    LockThread(pthread_mutex_t& m)
        :mutex(m)
    {
        pthread_mutex_lock(&mutex);
    }
    ~LockThread()
    {
        pthread_mutex_unlock(&mutex);
    }
    private:
        pthread_mutex_t& mutex;
};
class Consumer
{
    pthread_mutex_t     lock;
    pthread_cond_t      cond;
    std::vector<char>   unreadData;
    public:
    Consumer()
    {
        pthread_mutex_init(&lock,NULL);
        pthread_cond_init(&cond,NULL);
    }
    ~Consumer()
    {
        pthread_cond_destroy(&cond);
        pthread_mutex_destroy(&lock);
    }

    private:
        std::vector<char> read(size_t n)
        {
            LockThread  locker(lock);
            while (unreadData.size() < n)
            {
                // Must wait until we have n char.
                // This is a while loop because feed may not put enough in.

                // pthread_cond() releases the lock.
                // Thread will not be allowed to continue until
                // signal is called and this thread reacquires the lock.

                pthread_cond_wait(&cond,&lock);

                // Once released from the condition you will have re-aquired the lock.
                // Thus feed() must have exited and released the lock first.
            }

            /*
             * Not sure if this is exactly what you wanted.
             * But the data is copied out of the thread safe buffer
             * into something that can be returned.
             */
            std::vector<char>   result(n); // init result with size n
            std::copy(&unreadData[0],
                      &unreadData[n],
                      &result[0]);

            unreadData.erase(unreadData.begin(),
                             unreadData.begin() + n);
            return (result);
        }
public:
    void run()
    {
        read(10);
        read(4839);
        // etc
    }
    void feed(const std::vector<char> &more)
    {
        LockThread  locker(lock);

        // Once we acquire the lock we can safely modify the buffer.
        std::copy(more.begin(),more.end(),std::back_inserter(unreadData));

        // Only signal the thread if you have the lock
        // Otherwise race conditions happen.
        pthread_cond_signal(&cond);

        // destructor releases the lock and thus allows read thread to continue.
    }
};


int main()
{
    Consumer    c;
}

Answer 2

أنا أميل إلى استخدام ما يسميه "Syncronized الانتظار".أنا التفاف طابور العادي واستخدام إشارة الدرجة لكل قفل و جعل قراءة كتلة تماما كما تريدها:

#ifndef SYNCQUEUE_20061005_H_
#define SYNCQUEUE_20061005_H_

#include <queue>
#include "Semaphore.h"

// similar, but slightly simpler interface to std::queue
// this queue implementation will serialize pushes and pops
// and block on a pop while empty (as apposed to throwing an exception)
// it also locks as neccessary on insertion and removal to avoid race 
// conditions

template <class T, class C = std::deque<T> > class SyncQueue {
protected:
    std::queue<T, C>    m_Queue;
    Semaphore           m_Semaphore;
    Mutex               m_Mutex;

public:
    typedef typename std::queue<T, C>::value_type value_type;
    typedef typename std::queue<T, C>::size_type size_type;

    explicit SyncQueue(const C& a = C()) : m_Queue(a), m_Semaphore(0) {}

    bool empty() const              { return m_Queue.empty(); }
    size_type size() const          { return m_Queue.size(); }

    void push(const value_type& x);
    value_type pop();
};

template <class T, class C>
void SyncQueue<T, C>::push(const SyncQueue<T, C>::value_type &x) {
    // atomically push item
    m_Mutex.lock(); 
    m_Queue.push(x); 
    m_Mutex.unlock(); 

    // let blocking semaphore know another item has arrived
    m_Semaphore.v();
}

template <class T, class C>
typename SyncQueue<T, C>::value_type SyncQueue<T, C>::pop() {
    // block until we have at least one item
    m_Semaphore.p();

    // atomically read and pop front item
    m_Mutex.lock();
    value_type ret = m_Queue.front();
    m_Queue.pop();
    m_Mutex.unlock();

    return ret;
}

#endif

يمكنك تنفيذ الإشارات و mutexes المناسبة الأوليات في خيوط التنفيذ.

ملاحظة:هذا التطبيق هو مثال واحد من العناصر في قائمة الانتظار ، ولكن يمكن بسهولة التفاف هذا مع وظيفة المخازن النتائج حتى ن تم توفيرها.شيء مثل هذا لو كان طابور من حرف:

std::vector<char> func(int size) {
    std::vector<char> result;
    while(result.size() != size) {
        result.push_back(my_sync_queue.pop());
    }
    return result;
}

Answer 3

أنا رمي بعض شبه الزائفة رمز.هنا هي بلدي التعليقات:

1)كبيرة جدا من الحبوب من تأمين هنا.إذا كنت بحاجة إلى الوصول السريع, سوف تحتاج إلى إعادة النظر في هياكل البيانات الخاصة بك.المحكمة الخاصة بلبنان ليست threadsafe.

2)قفل منع حتى مزامنة يتيح ذلك من خلال.مزامنة هيكل هو أنه يتيح 1 الخيط من خلال ذلك في وقت واحد مع قفل/فتح آلية.لا حاجة الاقتراع أو نوعا من الاستثناء سقو هيكل.

3)هذا هو جميلة نحويا hacky قطع في المشكلة.أنا لا يجري الدقيق مع API أو C++ الجملة, ولكن أعتقد أنه يعطي لغويا الحل الصحيح.

4)تحرير ردا على التعليق.

class piper
{
pthread_mutex queuemutex;
pthread_mutex readymutex;
bool isReady; //init to false by constructor

//whatever else
};

piper::read()
{//whatever
pthread_mutex_lock(&queuemutex)
if(myqueue.size() >= n)
{ 
   return_queue_vector.push_back(/* you know what to do here */)

    pthread_mutex_lock(&readymutex)
    isReady = false;
    pthread_mutex_unlock(&readymutex)
}
pthread_mutex_unlock(&queuemutex)
}

piper::push_em_in()
{
//more whatever
pthread_mutex_lock(&queuemutex)
//push push push
if(myqueue.size() >= n)
{
    pthread_mutex_lock(&readymutex)
    isReady = true;
    pthread_mutex_unlock(&readymutex)
}
pthread_mutex_unlock(&queuemutex)
}

Answer 4

للمتعة فقط, هنا هو سريعة وقذرة التنفيذ باستخدام دفعة.ويستخدم pthreads تحت غطاء محرك السيارة على المنصات التي تدعم ذلك ، على ويندوز يستخدم ويندوز العمليات.

boost::mutex access;
boost::condition cond;

// consumer
data read()
{
  boost::mutex::scoped_lock lock(access);
  // this blocks until the data is ready
  cond.wait(lock);

  // queue is ready
  return data_from_queue();
}

// producer
void push(data)
{
  boost::mutex::scoped_lock lock(access);
  // add data to queue

  if (queue_has_enough_data())
    cond.notify_one();  
}

Answer 5

لمزيد من المرح هنا هو الإصدار النهائي.STL-أوتوماتيكية من دون سبب وجيه.:-)

#include <algorithm>
#include <deque>
#include <pthread.h>

template<typename T>
class MultithreadedReader {
    std::deque<T>   buffer;
    pthread_mutex_t moreDataMutex;
    pthread_cond_t  moreDataCond;

protected:
    template<typename OutputIterator>
    void read(size_t count, OutputIterator result) {
        pthread_mutex_lock(&moreDataMutex);

        while (buffer.size() < count) {
            pthread_cond_wait(&moreDataCond, &moreDataMutex);
        }
        std::copy(buffer.begin(), buffer.begin() + count, result);
        buffer.erase(buffer.begin(), buffer.begin() + count);

        pthread_mutex_unlock(&moreDataMutex);
    }

public:
    MultithreadedReader() {
        pthread_mutex_init(&moreDataMutex, 0);
        pthread_cond_init(&moreDataCond, 0);
    }

    ~MultithreadedReader() {
        pthread_cond_destroy(&moreDataCond);
        pthread_mutex_destroy(&moreDataMutex);
    }

    template<typename InputIterator>
    void feed(InputIterator first, InputIterator last) {
        pthread_mutex_lock(&moreDataMutex);

        buffer.insert(buffer.end(), first, last);
        pthread_cond_signal(&moreDataCond);

        pthread_mutex_unlock(&moreDataMutex);
    }
};

Answer 6

سطحي غير متزامن طوابير توفير تأمين النوم على القراءة فارغة طابور كنت تبحث عن.انظر http://library.gnome.org/devel/glib/2.20/glib-Asynchronous-Queues.html يمكنك الجمع بين لهم مع gthreads أو gthread حمامات.