packages feed

eventlog-socket (empty) → 0.1.0.0

raw patch · 7 files changed

+655/−0 lines, 7 filesdep +basesetup-changed

Dependencies added: base

Files

+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Revision history for eventlog-socket++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2018, Ben Gamari++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Ben Gamari nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ cbits/eventlog_socket.c view
@@ -0,0 +1,539 @@+// For POLLRDHUP+#define _GNU_SOURCE++#include <assert.h>+#include <stdbool.h>+#include <poll.h>+#include <string.h>+#include <stdlib.h>+#include <unistd.h>+#include <sys/types.h>+#include <sys/socket.h>+#include <sys/un.h>+#include <pthread.h>+#include <fcntl.h>++#include <Rts.h>++#include "eventlog_socket.h"++#define LISTEN_BACKLOG 5+#define POLL_LISTEN_TIMEOUT 10000+#define POLL_WRITE_TIMEOUT 1000++#ifndef POLLRDHUP+#define POLLRDHUP POLLHUP+#endif++// logging helper macros:+// - use PRINT_ERR to unconditionally log erroneous situations+// - otherwise use DEBUG_ERR+#define PRINT_ERR(...) fprintf(stderr, "ghc-eventlog-socket: " __VA_ARGS__)+#ifdef NDEBUG+#define DEBUG_ERR(fmt, ...)+#define DEBUG0_ERR(fmt)+#else+#define DEBUG_ERR(fmt, ...) fprintf(stderr, "ghc-eventlog-socket %s: " fmt, __func__, __VA_ARGS__)+#define DEBUG0_ERR(fmt) fprintf(stderr, "ghc-eventlog-socket %s: " fmt, __func__)+#endif++/*********************************************************************************+ * data definitions+ *********************************************************************************/+++struct write_buffer_item {+  uint8_t *orig; // original data pointer (which we free)+  uint8_t *data;+  size_t size; // invariant: size is not zero+  struct write_buffer_item *next;+};++// invariant: head and last are both NULL or both not NULL.+struct write_buffer {+  struct write_buffer_item *head;+  struct write_buffer_item *last;+};++/*********************************************************************************+ * globals+ *********************************************************************************/++/* This module is concurrent.+ * There are two thread(group)s:+ * 1. RTS+ * 2. worker spawned by open_socket+ */++// variables read and written by worker only:+static bool initialized = false;+static int listen_fd = -1;++// concurrency variables+static pthread_t listen_thread;+static pthread_cond_t new_conn_cond;+static pthread_mutex_t mutex;++// variables accessed by both threads.+// their access should be guarded by mutex.+//+// Note: RTS writes client_fd in writer_stop.+static volatile int client_fd = -1;+static struct write_buffer wt = {+  .head = NULL,+  .last = NULL,+};++/*********************************************************************************+ * write_buffer+ *********************************************************************************/++// push to the back.+void write_buffer_push(struct write_buffer *buf, uint8_t *data, size_t size) {+  DEBUG_ERR("%p, %lu\n", data, size);+  uint8_t *copy = malloc(size);+  memcpy(copy, data, size);++  struct write_buffer_item *item = malloc(sizeof(struct write_buffer_item));+  item->orig = copy;+  item->data = copy;+  item->size = size;+  item->next = NULL;++  struct write_buffer_item *last = buf->last;+  if (last == NULL) {+    assert(buf->head == NULL);++    buf->head = item;+    buf->last = item;+  } else {+    last->next = item;+    buf->last = item;+  }++  DEBUG_ERR("%p %p %p\n", buf, &wt, buf->head);+};++// pop from the front.+void write_buffer_pop(struct write_buffer *buf) {+  struct write_buffer_item *head = buf->head;+  if (head == NULL) {+    // buffer is empty: nothing to do.+    return;+  } else {+    buf->head = head->next;+    if (buf->last == head) {+      buf->last = NULL;+    }+    free(head->orig);+    free(head);+  }+}++// buf itself is not freed.+// it's safe to call write_buffer_free multiple times on the same buf.+void write_buffer_free(struct write_buffer *buf) {+  // not the most effecient implementation,+  // but should be obviously correct.+  while (buf->head) {+    write_buffer_pop(buf);+  }+}++/*********************************************************************************+ * EventLogWriter+ *********************************************************************************/++static void writer_init(void)+{+  // no-op+}++static void writer_enqueue(uint8_t *data, size_t size) {+  DEBUG_ERR("size: %p %lu\n", data, size);++  // TODO: check the size of the queue+  // if it's too big, we can start dropping blocks.++  // for now, we just push everythinb to the back of the buffer.+  write_buffer_push(&wt, data, size);++  DEBUG_ERR("wt.head = %p\n", wt.head);+}++static bool writer_write(void *eventlog, size_t size)+{+  DEBUG_ERR("size: %lu\n", size);+  pthread_mutex_lock(&mutex);+  int fd = client_fd;+  if (fd < 0) {+    goto exit;+  }++  DEBUG_ERR("client_fd = %d; wt.head = %p\n", fd, wt.head);++  if (wt.head != NULL) {+    // if there is stuff in queue already, we enqueue the current block.+    writer_enqueue(eventlog, size);+  } else {++    // and if there isn't, we can write immediately.+    int ret = write(fd, eventlog, size);+    DEBUG_ERR("write return %d\n", ret);++    if (ret == -1) {+      if (errno == EAGAIN || errno == EWOULDBLOCK) {+        // couldn't write anything, enqueue whole block+        writer_enqueue(eventlog, size);+        goto exit;+      } else if (errno == EPIPE) {+        // connection closed, simply exit+        goto exit;++      } else {+        PRINT_ERR("failed to write: %s\n", strerror(errno));+        goto exit;+      }+    } else {+      // we wrote something+      if (ret >= size) {+        // we wrote everything, nothing to do+        goto exit;+      } else {+        // we wrote only part of the buffer+        writer_enqueue(eventlog + ret, size - ret);+      }+    }+  }++exit:+  pthread_mutex_unlock(&mutex);+  return true;+}++static void writer_flush(void)+{+  // no-op+}++static void writer_stop(void)+{+  pthread_mutex_lock(&mutex);+  if (client_fd >= 0) {+    close(client_fd);+    client_fd = -1;+    write_buffer_free(&wt);+  }+  pthread_mutex_unlock(&mutex);+}++const EventLogWriter socket_writer = {+  .initEventLogWriter = writer_init,+  .writeEventLog = writer_write,+  .flushEventLog = writer_flush,+  .stopEventLogWriter = writer_stop+};++/*********************************************************************************+ * Main worker (in own thread)+ *********************************************************************************/++static void listen_iteration() {+  DEBUG0_ERR("enter");++  if (listen(listen_fd, LISTEN_BACKLOG) == -1) {+    PRINT_ERR("listen() failed: %s\n", strerror(errno));+    abort();+  }++  struct sockaddr_un remote;+  unsigned int len;++  struct pollfd pfd_accept = {+    .fd = listen_fd,+    .events = POLLIN,+    .revents = 0,+  };++  // poll until we can accept+  while (true) {+    int ret = poll(&pfd_accept, 1, POLL_LISTEN_TIMEOUT);+    if (ret ==  -1) {+      PRINT_ERR("poll() failed: %s\n", strerror(errno));+      return;+    } else if (ret == 0) {+      DEBUG0_ERR("accept poll timed out\n");+    } else {+      // got connection+      break;+    }+  }++  // accept+  int fd = accept(listen_fd, (struct sockaddr *) &remote, &len);+  if (fd == -1) {+    PRINT_ERR("accept failed: %s\n", strerror(errno));+  }++  // set socket into non-blocking mode+  int flags = fcntl(fd, F_GETFL);+  if (flags == -1) {+    PRINT_ERR("fnctl F_GETFL failed: %s\n", strerror(errno));+  }+  if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {+    PRINT_ERR("fnctl F_SETFL failed: %s\n", strerror(errno));+  }++  // we stop existing logging+  if (eventLogStatus() == EVENTLOG_RUNNING) {+    endEventLogging();+  }++  // we got client_id now.+  pthread_mutex_lock(&mutex);+  client_fd = fd;+  // Drop lock to allow initial batch of events to be written.+  pthread_mutex_unlock(&mutex);++  // start writing+  startEventLogging(&socket_writer);++  // Announce new connection+  pthread_cond_broadcast(&new_conn_cond);++  // we are done.+}++// nothing to write iteration.+//+// we poll only for whether the connection is closed.+static void nonwrite_iteration(int fd) {+  DEBUG_ERR("(%d)\n", fd);++  // Wait for socket to disconnect+  struct pollfd pfd = {+    .fd = fd,+    .events = POLLRDHUP,+    .revents = 0,+  };++  int ret = poll(&pfd, 1, POLL_WRITE_TIMEOUT);+  if (ret == -1 && errno != EAGAIN) {+    // error+    PRINT_ERR("poll() failed: %s\n", strerror(errno));+    return;+  } else if (ret == 0) {+    // timeout+    return;+  }++  // reset client_fd on RDHUP.+  if (pfd.revents | POLLRDHUP) {+    DEBUG_ERR("(%d) POLLRDHUP\n", fd);++    // reset client_fd+    pthread_mutex_lock(&mutex);+    // note: writer_stop may close the connection as well.+    client_fd = -1;+    write_buffer_free(&wt);+    pthread_mutex_unlock(&mutex);+    return;+  }++  // we don't stop logging,+  // write function will be no-op with negative client_fd+  //+  // Before setting new client_fd we will stop the logging,+  // and restart if afterwards, so the header is written+  // to the new connection.+}++// write iteration.+//+// we poll for both: can we write, and whether the connection is closed.+static void write_iteration(int fd) {+  DEBUG_ERR("(%d)\n", fd);++  // Wait for socket to disconnect+  struct pollfd pfd = {+    .fd = fd,+    .events = POLLOUT | POLLRDHUP,+    .revents = 0,+  };++  int ret = poll(&pfd, 1, POLL_WRITE_TIMEOUT);+  if (ret == -1 && errno != EAGAIN) {+    // error+    PRINT_ERR("poll() failed: %s\n", strerror(errno));+    return;+  } else if (ret == 0) {+    // timeout+    return;+  }++  // reset client_fd on RDHUP.+  if (pfd.revents & POLLHUP) {+    DEBUG_ERR("(%d) POLLRDHUP\n", fd);++    // reset client_fd+    pthread_mutex_lock(&mutex);+    assert(fd == client_fd);+    client_fd = -1;+    write_buffer_free(&wt);+    pthread_mutex_unlock(&mutex);+    return;+  }++  if (pfd.revents & POLLOUT) {+    DEBUG_ERR("(%d) POLLOUT\n", fd);++    pthread_mutex_lock(&mutex);+    while (wt.head) {+      struct write_buffer_item *item = wt.head;+      ret = write(client_fd, item->data, item->size);++      if (ret == -1) {+        if (errno == EAGAIN || errno == EWOULDBLOCK) {+          // couldn't write anything, shouldn't happend.+          // do nothing.+        } else if (errno == EPIPE) {+          client_fd = -1;+          write_buffer_free(&wt);+        } else {+          PRINT_ERR("failed to write: %s\n", strerror(errno));+        }++        // break out of the loop+        break;++      } else {+        // we wrote something+        if (ret >= item->size) {+          // we wrote whole element, try to write next element too+          write_buffer_pop(&wt);+          continue;+        } else {+          item->size -= ret;+          item->data += ret;+          break;+        }+      }+    }+    pthread_mutex_unlock(&mutex);+  }+}++static void iteration() {+  pthread_mutex_lock(&mutex);+  int fd = client_fd;+  bool empty = wt.head == NULL;+  DEBUG_ERR("fd = %d, wt.head = %p\n", fd, wt.head);+  pthread_mutex_unlock(&mutex);++  if (fd != -1) {+    if (empty) {+      nonwrite_iteration(fd);+    } else {+      write_iteration(fd);+    }+  } else {+    listen_iteration();+  }+}++/* Main loop of eventlog-socket own thread:+ * Currently it is two states:+ * - either we have connection, then we poll for writes (and drop of connection).+ * - or we don't have, then we poll for accept.+ */+static void *worker(void * _unused)+{+  while (true) {+    iteration();+  }++  return NULL; // unreachable+}++/*********************************************************************************+ * Initialization+ *********************************************************************************/++static void open_socket(const char *sock_path)+{+  DEBUG_ERR("enter: %s\n", sock_path);++  listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);++  struct sockaddr_un local;+  local.sun_family = AF_UNIX;+  strncpy(local.sun_path, sock_path, sizeof(local.sun_path) - 1);+  unlink(sock_path);+  if (bind(listen_fd, (struct sockaddr *) &local,+           sizeof(struct sockaddr_un)) == -1) {+    PRINT_ERR("failed to bind socket %s: %s\n", sock_path, strerror(errno));+    abort();+  }++  int ret = pthread_create(&listen_thread, NULL, worker, NULL);+  if (ret != 0) {+    PRINT_ERR("failed to spawn thread: %s\n", strerror(ret));+    abort();+  }+}+++/*********************************************************************************+ * Public interface+ *********************************************************************************/++void eventlog_socket_wait(void)+{+  pthread_mutex_lock(&mutex);+  while (client_fd == -1) {+    int ret = pthread_cond_wait(&new_conn_cond, &mutex);+    if (ret != 0) {+      PRINT_ERR("failed to wait on condition variable: %s\n", strerror(ret));+    }+  }+  pthread_mutex_unlock(&mutex);+}++void eventlog_socket_start(const char *sock_path, bool wait)+{+  if (!initialized) {+    pthread_mutex_init(&mutex, NULL);+    pthread_cond_init(&new_conn_cond, NULL);+    initialized = true;+  }++  if (!sock_path)+    return;++  if (eventLogStatus() == EVENTLOG_NOT_SUPPORTED) {+    PRINT_ERR("eventlog is not supported.\n");+    return;+  }++  // we stop existing logging+  if (eventLogStatus() == EVENTLOG_RUNNING) {+    endEventLogging();+  }++  // ... and restart with outer socket writer,+  // which is no-op so far.+  //+  // This trickery is to avoid+  //+  //     printAndClearEventLog: could not flush event log+  //+  // warning messages from showing up in stderr.+  startEventLogging(&socket_writer);++  open_socket(sock_path);+  if (wait) {+    DEBUG_ERR("ghc-eventlog-socket: Waiting for connection to %s...\n", sock_path);+    eventlog_socket_wait();+  }+}+
+ eventlog-socket.cabal view
@@ -0,0 +1,35 @@+cabal-version:      2.2+name:               eventlog-socket+version:            0.1.0.0+synopsis:           Stream GHC eventlog events to external processes+description:        Stream GHC eventlog events to external processes.+license:            BSD-3-Clause+license-file:       LICENSE+author:             Ben Gamari+maintainer:         ben@smart-cactus.org+copyright:          (c) 2018 Ben Gamari+category:           System+build-type:         Simple+Tested-With: GHC ==8.10.7 || ==9.0.2 || ==9.2.7 || ==9.4.4 || ==9.6.1+extra-source-files:+  ChangeLog.md+  include/eventlog_socket.h++source-repository head+  type:     git+  location: https://github.com/bgamari/ghc-eventlog-socket.git++library+  exposed-modules:  GHC.Eventlog.Socket+  other-extensions: CApiFFI++  -- Use base lower bound as proxy for GHC >= 8.10+  build-depends:    base >=4.14 && <5+  hs-source-dirs:   src+  default-language: Haskell2010+  c-sources:        cbits/eventlog_socket.c+  include-dirs:     include/++  -- comment me out while debugging+  -- https://github.com/haskell/cabal/issues/7635+  cc-options:      -DNDEBUG=1
+ include/eventlog_socket.h view
@@ -0,0 +1,7 @@+#ifndef EVENGLOG_SOCKET_H+#define EVENGLOG_SOCKET_H++void eventlog_socket_wait(void);+void eventlog_socket_start(const char *sock_path, bool wait);++#endif
+ src/GHC/Eventlog/Socket.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE CApiFFI #-}++-- |+-- Stream GHC eventlog events to external processes.+module GHC.Eventlog.Socket (+    startWait,+    start,+    wait,+) where++import Foreign.C+import Foreign.Ptr++-- | Start listening for eventlog connections, blocking until a client connects.+startWait :: FilePath  -- ^ File path to the unix domain socket to create.+          -> IO ()+startWait = c_start' True++-- | Start listening for eventlog connections.+start :: FilePath      -- ^ File path to the unix domain socket to create.+      -> IO ()+start = c_start' False++-- | Wait (block) until a client connects.+wait :: IO ()+wait = c_wait++c_start' :: Bool -> FilePath -> IO ()+c_start' block socketPath =+    withCString socketPath $ \socketPathCString ->+    c_start socketPathCString block++foreign import capi safe "eventlog_socket.h eventlog_socket_start"+    c_start :: CString -> Bool -> IO ()++foreign import capi safe "eventlog_socket.h eventlog_socket_wait"+    c_wait :: IO ()