Use x* memory functions to panic on OOM

This patch replaces all occurrences of memory related functions with
xmalloc, xcalloc and xrealloc, making them to panic, when we fail to
allocate memory for them.

Closes #3

Author: Serpentian

kafka/callbacks.c

@@ -25,14 +25,11 @@
 
 log_msg_t *
 new_log_msg(int level, const char *fac, const char *buf) {
-    log_msg_t *msg = malloc(sizeof(log_msg_t));
-    if (msg == NULL) {
-        return NULL;
-    }
+    log_msg_t *msg = xmalloc(sizeof(log_msg_t));
     msg->level = level;
-    msg->fac = malloc(sizeof(char) * strlen(fac) + 1);
+    msg->fac = xmalloc(sizeof(char) * strlen(fac) + 1);
     strcpy(msg->fac, fac);
-    msg->buf = malloc(sizeof(char) * strlen(buf) + 1);
+    msg->buf = xmalloc(sizeof(char) * strlen(buf) + 1);
     strcpy(msg->buf, buf);
     return msg;
 }
@@ -78,11 +75,9 @@ stats_callback(rd_kafka_t *rd_kafka, char *json, size_t json_len, void *opaque)
 
 error_msg_t *
 new_error_msg(int err, const char *reason) {
-    error_msg_t *msg = malloc(sizeof(error_msg_t));
-    if (msg == NULL)
-        return NULL;
+    error_msg_t *msg = xmalloc(sizeof(error_msg_t));
     msg->err = err;
-    msg->reason = malloc(sizeof(char) * strlen(reason) + 1);
+    msg->reason = xmalloc(sizeof(char) * strlen(reason) + 1);
     strcpy(msg->reason, reason);
     return msg;
 }
@@ -134,7 +129,7 @@ push_errors_cb_args(struct lua_State *L, const error_msg_t *msg)
 dr_msg_t *
 new_dr_msg(int dr_callback, int err) {
     dr_msg_t *dr_msg;
-    dr_msg = malloc(sizeof(dr_msg_t));
+    dr_msg = xmalloc(sizeof(dr_msg_t));
     dr_msg->dr_callback = dr_callback;
     dr_msg->err = err;
     return dr_msg;
@@ -165,11 +160,7 @@ msg_delivery_callback(rd_kafka_t *UNUSED(producer), const rd_kafka_message_t *ms
 
 rebalance_msg_t *
 new_rebalance_revoke_msg(rd_kafka_topic_partition_list_t *revoked) {
-    rebalance_msg_t *msg = malloc(sizeof(rebalance_msg_t));
-    if (msg == NULL) {
-        return NULL;
-    }
-
+    rebalance_msg_t *msg = xmalloc(sizeof(rebalance_msg_t));
     pthread_mutex_t lock;
     if (pthread_mutex_init(&lock, NULL) != 0) {
         free(msg);
@@ -193,11 +184,7 @@ new_rebalance_revoke_msg(rd_kafka_topic_partition_list_t *revoked) {
 
 rebalance_msg_t *
 new_rebalance_assign_msg(rd_kafka_topic_partition_list_t *assigned) {
-    rebalance_msg_t *msg = malloc(sizeof(rebalance_msg_t));
-    if (msg == NULL) {
-        return NULL;
-    }
-
+    rebalance_msg_t *msg = xmalloc(sizeof(rebalance_msg_t));
     pthread_mutex_t lock;
     if (pthread_mutex_init(&lock, NULL) != 0) {
         free(msg);
@@ -221,11 +208,7 @@ new_rebalance_assign_msg(rd_kafka_topic_partition_list_t *assigned) {
 
 rebalance_msg_t *
 new_rebalance_error_msg(rd_kafka_resp_err_t err) {
-    rebalance_msg_t *msg = malloc(sizeof(rebalance_msg_t));
-    if (msg == NULL) {
-        return NULL;
-    }
-
+    rebalance_msg_t *msg = xmalloc(sizeof(rebalance_msg_t));
     pthread_mutex_t lock;
     if (pthread_mutex_init(&lock, NULL) != 0) {
         free(msg);
@@ -326,7 +309,7 @@ rebalance_callback(rd_kafka_t *consumer, rd_kafka_resp_err_t err, rd_kafka_topic
 
 event_queues_t *
 new_event_queues() {
-    event_queues_t *event_queues = calloc(1, sizeof(event_queues_t));
+    event_queues_t *event_queues = xcalloc(1, sizeof(event_queues_t));
     for (int i = 0; i < MAX_QUEUE; i++)
         event_queues->cb_refs[i] = LUA_REFNIL;
     return event_queues;

kafka/consumer.c

@@ -90,10 +90,7 @@ consumer_poll_loop(void *arg) {
 
 static consumer_poller_t *
 new_consumer_poller(rd_kafka_t *rd_consumer) {
-    consumer_poller_t *poller = malloc(sizeof(consumer_poller_t));
-    if (poller == NULL)
-        return NULL;
-
+    consumer_poller_t *poller = xmalloc(sizeof(consumer_poller_t));
     poller->rd_consumer = rd_consumer;
     poller->should_stop = 0;
 
@@ -719,7 +716,7 @@ lua_create_consumer(struct lua_State *L) {
     consumer_poller_t *poller = new_consumer_poller(rd_consumer);
 
     consumer_t *consumer;
-    consumer = malloc(sizeof(consumer_t));
+    consumer = xmalloc(sizeof(consumer_t));
     consumer->rd_consumer = rd_consumer;
     consumer->topics = NULL;
     consumer->event_queues = event_queues;

kafka/consumer_msg.c

@@ -141,10 +141,7 @@ lua_consumer_msg_gc(struct lua_State *L) {
 msg_t *
 new_consumer_msg(rd_kafka_message_t *rd_message) {
     size_t message_size = sizeof(msg_t) + rd_message->len + rd_message->key_len;
-    msg_t *msg = calloc(message_size, 1);
-    if (msg == NULL)
-        return NULL;
-
+    msg_t *msg = xcalloc(message_size, 1);
     msg->topic = rd_message->rkt;
     msg->partition = rd_message->partition;
     msg->value = (char*)msg + sizeof(msg_t);

kafka/producer.c

@@ -53,10 +53,7 @@ producer_poll_loop(void *arg) {
 
 static producer_poller_t *
 new_producer_poller(rd_kafka_t *rd_producer) {
-    producer_poller_t *poller = malloc(sizeof(producer_poller_t));
-    if (poller == NULL)
-        return NULL;
-
+    producer_poller_t *poller = xmalloc(sizeof(producer_poller_t));
     poller->rd_producer = rd_producer;
     poller->should_stop = 0;
 
@@ -104,32 +101,24 @@ destroy_producer_poller(producer_poller_t *poller) {
 producer_topics_t *
 new_producer_topics(int32_t capacity) {
     rd_kafka_topic_t **elements;
-    elements = malloc(sizeof(rd_kafka_topic_t *) * capacity);
-    if (elements == NULL)
-        return NULL;
-
+    elements = xmalloc(sizeof(rd_kafka_topic_t *) * capacity);
     producer_topics_t *topics;
-    topics = malloc(sizeof(producer_topics_t));
+    topics = xmalloc(sizeof(producer_topics_t));
     topics->capacity = capacity;
     topics->count = 0;
     topics->elements = elements;
 
     return topics;
 }
 
-int
+void
 add_producer_topics(producer_topics_t *topics, rd_kafka_topic_t *element) {
     if (topics->count >= topics->capacity) {
-        rd_kafka_topic_t **new_elements = realloc(topics->elements, sizeof(rd_kafka_topic_t *) * topics->capacity * 2);
-        if (new_elements == NULL) {
-            printf("realloc failed to relloc rd_kafka_topic_t array.");
-            return 1;
-        }
+        rd_kafka_topic_t **new_elements = xrealloc(topics->elements, sizeof(rd_kafka_topic_t *) * topics->capacity * 2);
         topics->elements = new_elements;
         topics->capacity *= 2;
     }
     topics->elements[topics->count++] = element;
-    return 0;
 }
 
 static rd_kafka_topic_t *
@@ -312,10 +301,7 @@ lua_producer_produce(struct lua_State *L) {
             lua_pushstring(L, rd_kafka_err2str(rd_kafka_last_error()));
             goto error;
         }
-        if (add_producer_topics(producer->topics, rd_topic) != 0) {
-            lua_pushstring(L, "Unexpected error: failed to add new topic to topic list!");
-            goto error;
-        }
+        add_producer_topics(producer->topics, rd_topic);
     }
 
     rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;
@@ -572,7 +558,7 @@ lua_create_producer(struct lua_State *L) {
     producer_poller_t *poller = new_producer_poller(rd_producer);
 
     producer_t *producer;
-    producer = malloc(sizeof(producer_t));
+    producer = xmalloc(sizeof(producer_t));
     producer->rd_producer = rd_producer;
     producer->topics = new_producer_topics(256);
     producer->event_queues = event_queues;

kafka/producer.h

@@ -30,7 +30,7 @@ typedef struct {
 
 producer_topics_t *new_producer_topics(int32_t capacity);
 
-int
+void
 add_producer_topics(producer_topics_t *topics, rd_kafka_topic_t *element);
 
 void

kafka/queue.c

@@ -2,6 +2,7 @@
 #include <pthread.h>
 #include <unistd.h>
 
+#include <common.h>
 #include <queue.h>
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -54,18 +55,10 @@ queue_pop(queue_t *queue) {
  * @param value
  * @return
  */
-int
+void
 queue_lockfree_push(queue_t *queue, void *value) {
-    if (value == NULL || queue == NULL) {
-        return -1;
-    }
-
     queue_node_t *new_node;
-    new_node = malloc(sizeof(queue_node_t));
-    if (new_node == NULL) {
-        return -1;
-    }
-
+    new_node = xmalloc(sizeof(queue_node_t));
     new_node->value = value;
     new_node->next = NULL;
 
@@ -79,8 +72,6 @@ queue_lockfree_push(queue_t *queue, void *value) {
     }
 
     queue->count += 1;
-
-    return 0;
 }
 
 int
@@ -90,21 +81,15 @@ queue_push(queue_t *queue, void *value) {
     }
 
     pthread_mutex_lock(&queue->lock);
-
-    int output = queue_lockfree_push(queue, value);
-
+    queue_lockfree_push(queue, value);
     pthread_mutex_unlock(&queue->lock);
 
-    return output;
+    return 0;
 }
 
 queue_t *
 new_queue() {
-    queue_t *queue = malloc(sizeof(queue_t));
-    if (queue == NULL) {
-        return NULL;
-    }
-
+    queue_t *queue = xmalloc(sizeof(queue_t));
     pthread_mutex_t lock;
     if (pthread_mutex_init(&lock, NULL) != 0) {
         free(queue);

kafka/queue.h

@@ -41,7 +41,7 @@ queue_pop(queue_t *queue);
  * @param value
  * @return
  */
-int
+void
 queue_lockfree_push(queue_t *queue, void *value);
 
 int