async crash

test.c

#include <strings.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <uv.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>
#include <stdbool.h>

uv_loop_t *loop;
uv_async_t async;

long iter = 0;

// simple linked list of pending uv_async_t objects
typedef struct queue_node_s {
  uv_async_t async;
  struct queue_node_s *next;
  bool signalled;
} queue_node_t;

queue_node_t *queue_head = NULL;
pthread_mutex_t queue_mutex; // for queue_head.

void async_cb(uv_async_t *async);


// Add more work to the work thread
void push_work() {
  // This function must be run in a threadsafe way (while mutex is held)
  //pthread_mutex_lock(&queue_mutex);

  queue_node_t *node = malloc(sizeof(queue_node_t));
  node->next = queue_head;
  node->signalled = false;

  uv_async_init(loop, &node->async, async_cb);
  node->async.data = node;

  queue_head = node;
  //pthread_mutex_unlock(&queue_mutex);
}



void close_cb(uv_handle_t *async) {
  // Find the handle, free it and remove it from the list.
  pthread_mutex_lock(&queue_mutex);
  iter++;

  queue_node_t **prev_ptr = &queue_head;
  for (queue_node_t *node = queue_head; node != NULL; node = node->next) {
    if (&node->async == (uv_async_t *)async) {
      *prev_ptr = node->next;
      bzero(node, sizeof(queue_node_t));
      free(node);
      break;
    }

    prev_ptr = &node->next;
  }

  push_work();
  pthread_mutex_unlock(&queue_mutex);
  
  // And schedule more work!
}

void async_cb(uv_async_t *async) {
  //printf("async fired\n");
  uv_close((uv_handle_t *)async, close_cb);
}

void *net_thread(void *_unused) {
  printf("net thread running\n");
  while (1) {
    uv_async_t *async = NULL;

    pthread_mutex_lock(&queue_mutex);

    // Find the first item in the queue which hasn't been signalled yet
    for (queue_node_t *node = queue_head; node != NULL; node = node->next) {
      if (!node->signalled) {
        async = &node->async;
        node->signalled = true;
        break;
      }
    }

    pthread_mutex_unlock(&queue_mutex);

    // ... And signal it.
    if (async != NULL) {
      uv_async_send(async);
    }
  }
  return NULL;
}

void timer_cb(uv_timer_t *timer) {
  printf("completed %ld iterations\n", iter);
}

int main() {
  pthread_mutex_init(&queue_mutex, NULL);

  loop = uv_default_loop();

  uv_timer_t timer_req;
  uv_timer_init(loop, &timer_req);
  uv_timer_start(&timer_req, timer_cb, 5000, 5000);

  push_work();
  push_work();
  push_work();


  pthread_t net_thread_ptr;
  pthread_create(&net_thread_ptr, NULL, &net_thread, NULL);

  uv_run(loop, UV_RUN_DEFAULT);

  pthread_join(net_thread_ptr, NULL);

  return 0;
}