0

Господа и дамы,

я ищу класс на C++ который бы реализовывал thread-safe блокирующую очередь. Под блокирующей я понимаю:

  • когда пытаемся что-то прочитать из пустой очереди поток блокируется, пока кто-то другой не положит что-то в неё;

Писать самому этот велосипед не хочется. На крайний случай можно использовать mqueue, но мне хочется именно на C++

Спасибо.

1 ответ 1

1

Действительно mqueue.h несколько усложнено что-ли.

Если интересно, вот простой пример на Си (C++ тоже компируется и работает) для Linux (проверял в ubuntu). Можно просто переделать под свою структуру данных. Если разместить очередь и данные в shared memory, то можно использовать и для процессов с общим предком.

/*  q1ps.c
  Программка делает очередь (linked list) из строк
  main кладет fgets() в очередь
  threads печатают из нее и немного спят
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#ifndef __cplusplus
#define __USE_GNU
#define __USE_UNIX98
#endif
#include <pthread.h>
#include <semaphore.h>
#include <errno.h>
#include <time.h>

#ifndef __cplusplus
extern pid_t gettid(void);
#else
extern "C" pid_t gettid(void);
#endif

#define fatal(msg) do { perror(msg); exit(-1); } while(0)

struct qelem {
  struct qelem *next;
  char         *str;
};

struct queue {
  sem_t           sem;
  pthread_mutex_t lock;
  struct qelem    *head, *tail;
};

struct queue *
getqueue ()
{
  struct queue *q = (struct queue *)calloc(sizeof(*q),1);

  sem_init(&q->sem,0,0);
  pthread_mutexattr_t mattr;
  pthread_mutexattr_init(&mattr);
  pthread_mutexattr_settype(&mattr, PTHREAD_MUTEX_ERRORCHECK);
  pthread_mutex_init(&q->lock,&mattr);

  return q;
}

void
enqueue (struct queue *q, char *str)
{
  struct qelem *e = (struct qelem *)malloc(sizeof(*e));
  e->str = strdup(str);
  e->next = NULL;
  pthread_mutex_lock(&q->lock);
  if (q->head == NULL)
    q->tail = q->head = e;
  else {
    q->tail->next = e;
    q->tail = e;
  }
  sem_post(&q->sem);
  pthread_mutex_unlock(&q->lock);  
}

char *
dequeue (struct queue *q)
{
  struct qelem *e;
  sem_wait(&q->sem);
  pthread_mutex_lock(&q->lock);
  e = q->head;
  q->head = e->next;
  pthread_mutex_unlock(&q->lock);  
  char *res = e->str;
  free(e);
  return res;
}

void *
priq (void *a)
{
  struct queue *q = (struct queue *)a;
  pid_t tid = gettid();

  printf ("Thread %ld ready\n",(long)tid);

  for (;;) {
    char *str = dequeue(q);
    int sec = rand()%5;
    printf ("Thread %ld get [%s] and sleep %d sec\n",
        (long)tid,str,sec);
    free(str);
    sleep(sec);
  }
}

main ()
{
  struct queue *q = getqueue();

  pthread_attr_t attr;
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

  int i;
  for (i = 0; i < 3; i++) {
    pthread_t th;
    if (pthread_create(&th,&attr,priq,(void *)q))
      fatal("pth_create");
  }

  char str[1000];
  printf ("Press\n");
  while (fgets(str,1000,stdin)) {
    str[strlen(str)-1] = 0;
    enqueue(q,str);
    printf ("Press\n");
  }

  printf ("End\n");
  exit (0);

}

Это функция для идентификатора потока

/*  gettid.c
  avp 2011
  there are no gettid() in Linux libc
  so make it
*/

#include <sys/types.h>
#include <sys/syscall.h>

pid_t
gettid()
{
  return syscall(SYS_gettid);
}

g++ q1ps.c gettid.o -pthread

Если что-то неясно - спрашивайте.

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими пользовательским соглашением, политикой конфиденциальности и политикой о куки

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.