Merge branch 'ph3-fastevent'

src/Makefile.am

@@ -34,6 +34,7 @@ noinst_HEADERS = \
     module.h \
     reportxml.h \
     listensocket.h \
+    fastevent.h \
     event.h \
     event_log.h \
     event_exec.h \
@@ -78,6 +79,7 @@ icecast_SOURCES = \
     module.c \
     reportxml.c \
     listensocket.c \
+    fastevent.c \
     format.c \
     format_ogg.c \
     format_mp3.c \

src/client.c

@@ -49,6 +49,7 @@
 #include "util.h"
 #include "acl.h"
 #include "listensocket.h"
+#include "fastevent.h"
 
 /* for ADMIN_COMMAND_ERROR */
 #include "admin.h"
@@ -100,6 +101,8 @@ int client_create(client_t **c_ptr, connection_t *con, http_parser_t *parser)
     client->write_to_client = format_generic_write_to_client;
     *c_ptr = client;
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_CREATE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     return ret;
 }
 
@@ -205,6 +208,8 @@ void client_destroy(client_t *client)
     if (client == NULL)
         return;
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_DESTROY, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     if (client->reuse != ICECAST_REUSE_CLOSE) {
         /* only reuse the client if we reached the body's EOF. */
         if (client_body_eof(client) == 1) {
@@ -289,6 +294,8 @@ int client_read_bytes(client_t *client, void *buf, unsigned len)
     if (bytes == -1 && client->con->error)
         ICECAST_LOG_DEBUG("reading from connection has failed");
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_READ, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_OBRD, client, buf, (size_t)len, (ssize_t)bytes);
+
     return bytes;
 }
 
@@ -386,6 +393,8 @@ void client_send_101(client_t *client, reuse_t reuse)
     client->respcode = 101;
     client->refbuf->len = strlen(client->refbuf->data);
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_SEND_RESPONSE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     fserve_add_client(client, NULL);
 }
 
@@ -409,6 +418,8 @@ void client_send_204(client_t *client)
     client->respcode = 204;
     client->refbuf->len = strlen(client->refbuf->data);
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_SEND_RESPONSE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     fserve_add_client(client, NULL);
 }
 
@@ -439,6 +450,8 @@ void client_send_426(client_t *client, reuse_t reuse)
 
     client->reuse = ICECAST_REUSE_KEEPALIVE;
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_SEND_RESPONSE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     fserve_add_client(client, NULL);
 }
 
@@ -450,6 +463,10 @@ static inline void client_send_500(client_t *client, const char *message)
     const ssize_t header_len = sizeof(header) - 1;
     ssize_t ret;
 
+    client->respcode = 500;
+    client->refbuf->len = 0;
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_SEND_RESPONSE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     ret = client_send_bytes(client, header, header_len);
 
     /* only send message if we have one AND if header could have transmitted completly */
@@ -569,6 +586,7 @@ void client_send_reportxml(client_t *client, reportxml_t *report, document_domai
         client->refbuf->len = ret;
         xmlFree(buff);
         client->respcode = status;
+        fastevent_emit(FASTEVENT_TYPE_CLIENT_SEND_RESPONSE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
         fserve_add_client (client, NULL);
     } else {
         char *fullpath_xslt_template;
@@ -597,6 +615,7 @@ void client_send_reportxml(client_t *client, reportxml_t *report, document_domai
         config_release_config();
 
         ICECAST_LOG_DEBUG("Sending XSLT (%s)", fullpath_xslt_template);
+        fastevent_emit(FASTEVENT_TYPE_CLIENT_SEND_RESPONSE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
         xslt_transform(doc, fullpath_xslt_template, client, status);
         free(fullpath_xslt_template);
     }
@@ -632,6 +651,8 @@ int client_send_bytes(client_t *client, const void *buf, unsigned len)
     if (client->con->error)
         ICECAST_LOG_DEBUG("Client connection died");
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_WRITE, FASTEVENT_FLAG_NONE, FASTEVENT_DATATYPE_OBRD, client, buf, (size_t)len, (ssize_t)ret);
+
     return ret;
 }
 
@@ -667,6 +688,8 @@ ssize_t client_body_read(client_t *client, void *buf, size_t len)
         client->request_body_read += ret;
     }
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_READ_BODY, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_OBRD, client, buf, len, ret);
+
     return ret;
 }
 

src/connection.c

@@ -60,6 +60,7 @@
 #include "acl.h"
 #include "refobject.h"
 #include "listensocket.h"
+#include "fastevent.h"
 
 #define CATMODULE "connection"
 
@@ -209,6 +210,7 @@ static int connection_send_tls(connection_t *con, const void *buf, size_t len)
     } else {
         con->sent_bytes += bytes;
     }
+
     return bytes;
 }
 #else
@@ -245,6 +247,7 @@ static int connection_send(connection_t *con, const void *buf, size_t len)
     } else {
         con->sent_bytes += bytes;
     }
+
     return bytes;
 }
 
@@ -271,6 +274,8 @@ connection_t *connection_create(sock_t sock, listensocket_t *listensocket_real,
         con->send       = connection_send;
     }
 
+    fastevent_emit(FASTEVENT_TYPE_CONNECTION_CREATE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CONNECTION, con);
+
     return con;
 }
 
@@ -299,10 +304,14 @@ void connection_uses_tls(connection_t *con)
 
 ssize_t connection_send_bytes(connection_t *con, const void *buf, size_t len)
 {
-    return con->send(con, buf, len);
+    ssize_t ret = con->send(con, buf, len);
+
+    fastevent_emit(FASTEVENT_TYPE_CONNECTION_WRITE, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_OBRD, con, buf, len, ret);
+
+    return ret;
 }
 
-ssize_t connection_read_bytes(connection_t *con, void *buf, size_t len)
+static inline ssize_t connection_read_bytes_real(connection_t *con, void *buf, size_t len)
 {
     ssize_t done = 0;
     ssize_t ret;
@@ -344,10 +353,21 @@ ssize_t connection_read_bytes(connection_t *con, void *buf, size_t len)
     return done + ret;
 }
 
+ssize_t connection_read_bytes(connection_t *con, void *buf, size_t len)
+{
+    ssize_t ret = connection_read_bytes_real(con, buf, len);
+
+    fastevent_emit(FASTEVENT_TYPE_CONNECTION_READ, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_OBRD, con, buf, len, ret);
+
+    return ret;
+}
+
 int connection_read_put_back(connection_t *con, const void *buf, size_t len)
 {
     void *n;
 
+    fastevent_emit(FASTEVENT_TYPE_CONNECTION_PUTBACK, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_OBR, con, buf, len);
+
     if (con->readbufferlen) {
         n = realloc(con->readbuffer, con->readbufferlen + len);
         if (!n)
@@ -1257,6 +1277,8 @@ static void _handle_authed_client(client_t *client, void *uri, auth_result resul
     auth_stack_release(client->authstack);
     client->authstack = NULL;
 
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_AUTHED, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
+
     if (result != AUTH_OK) {
         client_send_error_by_id(client, ICECAST_ERROR_GEN_CLIENT_NEEDS_TO_AUTHENTICATE);
         free(uri);
@@ -1397,6 +1419,7 @@ static void _handle_authentication_mount_normal(client_t *client, char *uri)
 
 static void _handle_authentication(client_t *client, char *uri)
 {
+    fastevent_emit(FASTEVENT_TYPE_CLIENT_READY_FOR_AUTH, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CLIENT, client);
     _handle_authentication_mount_normal(client, uri);
 }
 
@@ -1650,6 +1673,8 @@ void connection_close(connection_t *con)
     if (!con)
         return;
 
+    fastevent_emit(FASTEVENT_TYPE_CONNECTION_DESTROY, FASTEVENT_FLAG_MODIFICATION_ALLOWED, FASTEVENT_DATATYPE_CONNECTION, con);
+
     tls_unref(con->tls);
     if (con->sock != -1) /* TODO: do not use magic */
         sock_close(con->sock);

src/event.c

@@ -18,6 +18,7 @@
 #include "event_log.h"
 #include "event_exec.h"
 #include "event_url.h"
+#include "fastevent.h"
 #include "logging.h"
 #include "admin.h"
 #include "connection.h"
@@ -334,6 +335,7 @@ void event_registration_push(event_registration_t **er, event_registration_t *ta
 
 /* event signaling */
 void event_emit(event_t *event) {
+    fastevent_emit(FASTEVENT_TYPE_SLOWEVENT, FASTEVENT_FLAG_NONE, FASTEVENT_DATATYPE_EVENT, event);
     thread_mutex_lock(&event_lock);
     event_push(&event_queue, event);
     thread_mutex_unlock(&event_lock);
@@ -371,11 +373,13 @@ void event_emit_clientevent(const char *trigger, client_t *client, const char *u
      * We do this before inserting all the data into the object to avoid
      * all the strdups() and stuff in case they aren't needed.
      */
+#ifndef FASTEVENT_ENABLED
     if (event->reglist[0] == NULL) {
         /* we have no registrations, drop this event. */
         event_release(event);
         return;
     }
+#endif
 
     if (client) {
         const char *tmp;

src/fastevent.c

+/* Icecast
+ *
+ * This program is distributed under the GNU General Public License, version 2.
+ * A copy of this license is included with this source.
+ *
+ * Copyright 2018,      Philipp "ph3-der-loewe" Schafft <lion@lion.leolix.org>,
+ */
+
+/**
+ * Special fast event functions
+ */
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <stdlib.h>
+#include <string.h>
+
+#include "common/thread/thread.h"
+
+#include "fastevent.h"
+
+#include "logging.h"
+#define CATMODULE "fastevent"
+
+#ifdef FASTEVENT_ENABLED
+
+struct registration {
+    refobject_base_t __base;
+
+    fastevent_type_t type;
+    fastevent_cb_t cb;
+    fastevent_freecb_t freecb;
+    void *userdata;
+};
+
+struct eventrow {
+    size_t length;
+    size_t used;
+    struct registration **registrations;
+};
+
+static struct eventrow fastevent_registrations[FASTEVENT_TYPE__END];
+static rwlock_t fastevent_lock;
+
+static inline struct eventrow * __get_row(fastevent_type_t type)
+{
+    size_t idx = type;
+
+    if (idx >= FASTEVENT_TYPE__END)
+        return NULL;
+
+    return &(fastevent_registrations[idx]);
+}
+
+static int __add_to_row(struct eventrow * row, struct registration *registration)
+{
+    struct registration **n;
+
+    if (row == NULL)
+        return -1;
+
+    /* Check if we need to reallocate row space */
+    if (row->length == row->used) {
+        n = realloc(row->registrations, sizeof(*n)*(row->length + 4));
+        if (n == NULL) {
+            ICECAST_LOG_ERROR("Can not allocate row space.");
+            return -1;
+        }
+
+        row->registrations = n;
+        row->length += 4;
+    }
+
+    row->registrations[row->used++] = registration;
+    return 0;
+}
+
+static int __remove_from_row(struct eventrow * row, struct registration *registration)
+{
+    size_t i;
+
+    if (row == NULL)
+        return -1;
+
+    for (i = 0; i < row->used; i++) {
+        if (row->registrations[i] == registration) {
+            memmove(&(row->registrations[i]), &(row->registrations[i+1]), sizeof(*(row->registrations))*(row->used - i - 1));
+            row->used--;
+            return 0;
+        }
+    }
+
+    return -1;
+}
+
+
+static void __unregister(refobject_t self, void **userdata)
+{
+    struct registration *registration = REFOBJECT_TO_TYPE(self, struct registration *);
+    struct eventrow * row;
+
+    (void)userdata;
+
+    thread_rwlock_wlock(&fastevent_lock);
+    row = __get_row(registration->type);
+    if (__remove_from_row(row, registration) != 0) {
+        ICECAST_LOG_ERROR("Can not remove fast event from row. BUG.");
+    }
+    thread_rwlock_unlock(&fastevent_lock);
+
+    if (registration->freecb)
+        registration->freecb(&(registration->userdata));
+
+    if (registration->userdata != NULL)
+        free(registration->userdata);
+}
+
+int fastevent_initialize(void)
+{
+    thread_rwlock_create(&fastevent_lock);
+    return 0;
+}
+
+int fastevent_shutdown(void)
+{
+    size_t i;
+
+    thread_rwlock_wlock(&fastevent_lock);
+    for (i = 0; i < FASTEVENT_TYPE__END; i++) {
+        if (fastevent_registrations[i].used) {
+            ICECAST_LOG_ERROR("Subsystem shutdown but elements still in use. BUG.");
+            continue;
+        }
+
+        free(fastevent_registrations[i].registrations);
+        fastevent_registrations[i].registrations = NULL;
+    }
+    thread_rwlock_unlock(&fastevent_lock);
+    thread_rwlock_destroy(&fastevent_lock);
+
+    return 0;
+}
+
+refobject_t fastevent_register(fastevent_type_t type, fastevent_cb_t cb, fastevent_freecb_t freecb, void *userdata)
+{
+    struct eventrow * row;
+    struct registration *registration;
+    refobject_t ret;
+
+    if (cb == NULL)
+        return REFOBJECT_NULL;
+
+    thread_rwlock_wlock(&fastevent_lock);
+    row = __get_row(type);
+
+    if (row == NULL) {
+        thread_rwlock_unlock(&fastevent_lock);
+        return REFOBJECT_NULL;
+    }
+
+    ret = refobject_new(sizeof(struct registration), __unregister, NULL, NULL, NULL);
+
+    if (REFOBJECT_IS_NULL(ret)) {
+        thread_rwlock_unlock(&fastevent_lock);
+        return REFOBJECT_NULL;
+    }
+
+    registration = REFOBJECT_TO_TYPE(ret, struct registration *);
+
+    registration->type = type;
+    registration->cb = cb;
+    registration->freecb = freecb;
+    registration->userdata = userdata;
+
+    if (__add_to_row(row, registration) != 0) {
+        thread_rwlock_unlock(&fastevent_lock);
+        refobject_unref(ret);
+        return REFOBJECT_NULL;
+    }
+
+    thread_rwlock_unlock(&fastevent_lock);
+    return ret;
+}
+
+void fastevent_emit(fastevent_type_t type, fastevent_flag_t flags, fastevent_datatype_t datatype, ...)
+{
+    struct eventrow * row;
+    va_list ap, apx;
+    size_t i;
+
+    ICECAST_LOG_DEBUG("event: type=%i, flags=%i, datatype=%i, ...", (int)type, (int)flags, (int)datatype);
+
+    thread_rwlock_rlock(&fastevent_lock);
+    row = __get_row(type);
+    if (row == NULL || row->used == 0) {
+        thread_rwlock_unlock(&fastevent_lock);
+        return;
+    }
+
+    va_start(ap, datatype);
+
+    for (i = 0; i < row->used; i++) {
+        va_copy(apx, ap);
+        row->registrations[i]->cb(row->registrations[i]->userdata, type, flags, datatype, apx);
+        va_end(apx);
+    }
+
+    thread_rwlock_unlock(&fastevent_lock);
+
+    va_end(ap);
+}
+#endif

src/fastevent.h

+/* Icecast
+ *
+ * This program is distributed under the GNU General Public License, version 2.
+ * A copy of this license is included with this source.
+ *
+ * Copyright 2018,      Philipp "ph3-der-loewe" Schafft <lion@lion.leolix.org>,
+ */
+
+#ifndef __FASTEVENT_H__
+#define __FASTEVENT_H__
+
+/* Add all conditions when to enable fast events here. */
+#if 1
+#define FASTEVENT_ENABLED
+#endif
+
+#include <stdarg.h>
+#include <refobject.h>
+
+typedef enum {
+    FASTEVENT_TYPE_SLOWEVENT = 0,
+    FASTEVENT_TYPE_CONNECTION_CREATE,
+    FASTEVENT_TYPE_CONNECTION_DESTROY,
+    FASTEVENT_TYPE_CONNECTION_READ,
+    FASTEVENT_TYPE_CONNECTION_PUTBACK,
+    FASTEVENT_TYPE_CONNECTION_WRITE,
+    FASTEVENT_TYPE_CLIENT_CREATE,
+    FASTEVENT_TYPE_CLIENT_DESTROY,
+    FASTEVENT_TYPE_CLIENT_READ,
+    FASTEVENT_TYPE_CLIENT_WRITE,
+    FASTEVENT_TYPE_CLIENT_READ_BODY,
+    FASTEVENT_TYPE_CLIENT_READY_FOR_AUTH,
+    FASTEVENT_TYPE_CLIENT_AUTHED,
+    FASTEVENT_TYPE_CLIENT_SEND_RESPONSE,
+    FASTEVENT_TYPE__END /* must be last element */
+} fastevent_type_t;
+
+typedef enum {
+    FASTEVENT_DATATYPE_NONE = 0,
+    FASTEVENT_DATATYPE_EVENT,
+    FASTEVENT_DATATYPE_CLIENT,
+    FASTEVENT_DATATYPE_CONNECTION,
+    FASTEVENT_DATATYPE_OBR,             /* Object, const void *Buffer, size_t Request_length */
+    FASTEVENT_DATATYPE_OBRD             /* Object, const void *Buffer, size_t Request_length, ssize_t Done_length */
+} fastevent_datatype_t;
+
+typedef int fastevent_flag_t;
+#define FASTEVENT_FLAG_NONE                     ((fastevent_flag_t)0x0000)
+#define FASTEVENT_FLAG_MODIFICATION_ALLOWED     ((fastevent_flag_t)0x0001)
+
+typedef void (*fastevent_cb_t)(const void *userdata, fastevent_type_t type, fastevent_flag_t flags, fastevent_datatype_t datatype, va_list ap);
+typedef void (*fastevent_freecb_t)(void **userdata);
+
+#ifdef FASTEVENT_ENABLED
+int fastevent_initialize(void);
+int fastevent_shutdown(void);
+refobject_t fastevent_register(fastevent_type_t type, fastevent_cb_t cb, fastevent_freecb_t freecb, void *userdata);
+void fastevent_emit(fastevent_type_t type, fastevent_flag_t flags, fastevent_datatype_t datatype, ...);
+#else
+#define fastevent_initialize() 0
+#define fastevent_shutdown() 0
+#define fastevent_register(type,cb,freecb,userdata) REFOBJECT_NULL
+#define fastevent_emit(type,flags,datatype,...)
+#endif
+
+#endif

src/main.c

@@ -78,6 +78,7 @@
 #include "auth.h"
 #include "event.h"
 #include "listensocket.h"
+#include "fastevent.h"
 
 #include <libxml/xmlmemory.h>
 
@@ -119,11 +120,35 @@ static void _stop_logging(void)
     log_close(playlistlog);
 }
 
+#ifndef FASTEVENT_ENABLED
+static void __fastevent_cb(const void *userdata, fastevent_type_t type, fastevent_flag_t flags, fastevent_datatype_t datatype, va_list ap)
+{
+    event_t *event;
+
+    if (datatype != FASTEVENT_DATATYPE_EVENT)
+        return;
+
+    event = va_arg(ap, event_t*);
+
+    if (event == NULL) {
+        ICECAST_LOG_DEBUG("event=%p", event);
+    } else {
+        ICECAST_LOG_DEBUG("event=%p{.trigger='%s', ...}", event, event->trigger);
+    }
+}
+
+static refobject_t fastevent_reg;
+#endif
+
 static void initialize_subsystems(void)
 {
     log_initialize();
     thread_initialize();
     global_initialize();
+#ifndef FASTEVENT_ENABLED
+    fastevent_initialize();
+    fastevent_reg = fastevent_register(FASTEVENT_TYPE_SLOWEVENT, __fastevent_cb, NULL, NULL);
+#endif
     sock_initialize();
     resolver_initialize();
     config_initialize();
@@ -152,6 +177,10 @@ static void shutdown_subsystems(void)
     config_shutdown();
     resolver_shutdown();
     sock_shutdown();
+#ifndef FASTEVENT_ENABLED
+    refobject_unref(fastevent_reg);
+    fastevent_shutdown();
+#endif
     global_shutdown();
     thread_shutdown();