Read input from another thread

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>

typedef enum { USER_INPUT, COMMAND } EventType;

typedef struct event_t
{
  EventType type;
  void *data;
  struct event_t * next;
} event_t;

typedef struct
{
  pthread_mutex_t mutex;
  pthread_cond_t cond;
  event_t * head;
  event_t * tail;
} event_queue_t;

pthread_t user_input_thread;
pthread_t async_commands_thread;
event_queue_t event_queue;
char *current_input;

void queue_init()
{
  pthread_mutex_init(&event_queue.mutex, NULL);
  pthread_cond_init(&event_queue.cond, NULL);
  event_queue.head = NULL;
  event_queue.tail = NULL;
}


void queue_add(EventType type, void *data)
{
  event_t *ev = (event_t *)malloc(sizeof(event_t));
  ev->type = type;
  ev->data = data;
  ev->next = NULL;

  pthread_mutex_lock(&event_queue.mutex);

  if (event_queue.head == NULL) {
    event_queue.head = ev;
    // queue was empty and consequently the main thread was waiting, so wake it
    // up to continue after the lock is released
    pthread_cond_broadcast(&event_queue.cond);
  } else {
    if (event_queue.tail != NULL) {
      event_queue.tail->next = ev;
    } else {
      event_queue.head->next = ev;
    }
    event_queue.tail = ev;
  }

  pthread_mutex_unlock(&event_queue.mutex);

}

event_t * queue_remove()
{
  event_t *rv;
  pthread_mutex_lock(&event_queue.mutex);

  if (event_queue.head == NULL) {
    // queue is empty, temporarily release the lock and wait for events
    pthread_cond_wait(&event_queue.cond, &event_queue.mutex);
  }

  rv = event_queue.head;
  event_queue.head = rv->next;

  pthread_mutex_unlock(&event_queue.mutex);

  return rv;
}


// This will put a string that will be evaluated as a vim command in the next
// main_loop iteration.
void enqueue_command(char *command_str)
{
  queue_add(COMMAND, command_str);
}


void * generate_async_commands()
{
  while (1)
  {
    sleep(5);
    char *command_str = (char *)malloc(1024);
    strcpy(command_str, "[SOME VIMSCRIPT CODE TO EXECUTE]");
    enqueue_command(command_str);
  }
}


void * vgetc(void *arg)
{
  while (1)
  {
    char *data = (char *)malloc(1024);
    scanf("%s", data);
    queue_add(USER_INPUT, data);
  }

  return NULL;
}


int next_event()
{
  event_t *next = queue_remove();
  switch (next->type)
  {
    case USER_INPUT:
      current_input = next->data;
      free(next);
      return 0;
    case COMMAND:
      printf("Received command: '%s'\n", (char *)next->data);
      free(next->data);
      free(next);
      return 1;
  }
}


void main_loop()
{
  pthread_attr_t attr;
  pthread_attr_init(&attr);

  // Create a dedicated thread for reading user input
  pthread_create(&user_input_thread, &attr, &vgetc, NULL);

  // Create a background thread to illustrate communication from plugins
  pthread_create(&async_commands_thread, &attr, &generate_async_commands,
      NULL);

  while (1) {
    if (next_event()) continue;
    printf("User input '%s'\n", current_input);
    free(current_input);
  }
}


int main()
{
  main_loop();
  return 0;
}