[DRAFT] Share circuit breakers between workers

.gitignore

@@ -4,6 +4,8 @@
 /tmp
 /pkg
 *.gem
+*.log
+*.orig
 /html/
 Gemfile.lock
 vendor/
@@ -18,3 +20,5 @@ nohup.out
 
 # IntelliJ/RubyMine/CLion project files
 .idea
+CMakeLists.txt
+cmake-build-debug

.travis.yml

@@ -13,15 +13,32 @@ before_install:
 services:
   - docker
 
+gemfile:
+  - Gemfile
+  - gemfiles/hiredis-0-6.gemfile
+  - gemfiles/mysql2-0-4-10.gemfile
+  - gemfiles/mysql2-0-5-0.gemfile
+
+env:
+  - SEMIAN_CIRCUIT_BREAKER_IMPL=worker
+  - SEMIAN_CIRCUIT_BREAKER_IMPL=host
+
 matrix:
-  include:
-  - env: GEMFILE=gemfiles/hiredis-0-6.gemfile
-  - env: GEMFILE=gemfiles/mysql2-0-4-10.gemfile
-  - env: GEMFILE=gemfiles/mysql2-0-5-0.gemfile
-  - env: GEMFILE=Gemfile
+  exclude:
+  - gemfile: gemfiles/hiredis-0-6.gemfile
+    env: SEMIAN_CIRCUIT_BREAKER_IMPL=host
+  - gemfile: gemfiles/mysql2-0-4-10.gemfile
+    env: SEMIAN_CIRCUIT_BREAKER_IMPL=host
+  - gemfile: gemfiles/mysql2-0-5-0.gemfile
+    env: SEMIAN_CIRCUIT_BREAKER_IMPL=host
 
 script:
-  - docker build --build-arg BUNDLE_GEMFILE="$GEMFILE" -t shopify/semian-ci:latest -f dockerfiles/semian-ci .
+  - |
+    docker build \
+    --build-arg BUNDLE_GEMFILE="$BUNDLE_GEMFILE" \
+    --build-arg SEMIAN_CIRCUIT_BREAKER_IMPL="$SEMIAN_CIRCUIT_BREAKER_IMPL" \
+    -t shopify/semian-ci:latest \
+    -f dockerfiles/semian-ci .
   - travis_retry docker-compose -f docker-compose.ci.yml up --force-recreate --exit-code-from semian
 
 notifications:

README.md

@@ -448,6 +448,33 @@ There are three configuration parameters for circuit breakers in Semian:
 * **success_threshold**. The amount of successes on the circuit until closing it
   again, that is to start accepting all requests to the circuit.
 * **half_open_resource_timeout**. Timeout for the resource in seconds when the circuit is half-open (only supported for MySQL and Net::HTTP).
+* **scale_factor**. When using [host-based circuits](#host-based-circuits), the
+  scaling factor to determine how to scale `error_threshold * num_workers` to
+  achieve faster circuit opens.
+
+#### Host-Based Circuits
+
+On systems with [SysV support][sysv], we can share circuit error information
+between processes on a server. This means that those processes can effectively
+share information between each other about resource health, leading to faster,
+more efficient opening of circuits.
+
+As an example, imagine a system with _N_ processes on a host with an error
+threshold of _E_, and a client timeout of _T_. By default, the system needs to
+see _N * E_ errors to open the circuit. We can reduce this by using the
+`scale_factor` configuration parameter. If we set `scale_factor` to _1 / N_,
+the total number of errors we'd need to see server-wide is still _E_. In
+this configuration, we can reduce the time-to-open for a circuit from _E * T_
+to simply _T_ (provided that _N_ is greater than _E_).
+
+You should run a simulation with your workloads to determine an efficient
+scaling factor that will produce a time-to-open reduction but isn't too
+sensitive.
+
+The circuit breaker implementation is based on the environment variable
+`SEMIAN_CIRCUIT_BREAKER_IMPL`. Set it to `worker` to disable sharing circuit
+state and `host` to enable host-based circuits. The default is `worker` for
+backward compatibility.
 
 ### Bulkheading
 

dockerfiles/semian-ci

@@ -7,7 +7,10 @@ RUN apt-get update \
 
 RUN gem install bundler
 
-WORKDIR /app
+WORKDIR /home/travis/build/Shopify/semian
 COPY . .
 ARG BUNDLE_GEMFILE
 ENV BUNDLE_GEMFILE="${BUNDLE_GEMFILE}"
+
+ARG SEMIAN_CIRCUIT_BREAKER_IMPL
+ENV SEMIAN_CIRCUIT_BREAKER_IMPL="${SEMIAN_CIRCUIT_BREAKER_IMPL}"

ext/semian/circuit_breaker.c

@@ -0,0 +1,67 @@
+#include "circuit_breaker.h"
+
+#include "sysv_semaphores.h"
+#include "sysv_shared_memory.h"
+#include "types.h"
+#include "util.h"
+
+void
+semian_circuit_breaker_free(void* ptr)
+{
+  semian_circuit_breaker_t* res = (semian_circuit_breaker_t*)ptr;
+  free_shared_memory(res->shmem);
+}
+
+size_t
+semian_circuit_breaker_size(const void* ptr)
+{
+  return sizeof(semian_circuit_breaker_t);
+}
+
+static const rb_data_type_t semian_circuit_breaker_type = {
+  .wrap_struct_name = "semian_circuit_breaker",
+  .function = {
+    .dmark = NULL,
+    .dfree = semian_circuit_breaker_free,
+    .dsize = semian_circuit_breaker_size,
+  },
+  .data = NULL,
+  .flags = RUBY_TYPED_FREE_IMMEDIATELY,
+};
+
+void
+Init_CircuitBreaker() {
+  dprintf("Init_CircuitBreaker");
+
+  VALUE cSemian = rb_const_get(rb_cObject, rb_intern("Semian"));
+  VALUE cCircuitBreaker = rb_const_get(cSemian, rb_intern("CircuitBreaker"));
+
+  rb_define_alloc_func(cCircuitBreaker, semian_circuit_breaker_alloc);
+  rb_define_method(cCircuitBreaker, "initialize_circuit_breaker", semian_circuit_breaker_initialize, 2);
+}
+
+VALUE
+semian_circuit_breaker_alloc(VALUE klass)
+{
+  semian_circuit_breaker_t *res;
+  VALUE obj = TypedData_Make_Struct(klass, semian_circuit_breaker_t, &semian_circuit_breaker_type, res);
+  return obj;
+}
+
+VALUE
+semian_circuit_breaker_initialize(VALUE self, VALUE name)
+{
+  semian_circuit_breaker_t *res = NULL;
+  TypedData_Get_Struct(self, semian_circuit_breaker_t, &semian_circuit_breaker_type, res);
+
+  char buffer[1024];
+  strcpy(buffer, to_s(name));
+  strcat(buffer, "_circuit_breaker");
+  res->key = generate_key(buffer);
+
+  dprintf("Initializing circuit breaker '%s' (key: %lu)", buffer, res->key);
+  res->sem_id = initialize_single_semaphore(res->key, SEM_DEFAULT_PERMISSIONS);
+  res->shmem = get_or_create_shared_memory(res->key, NULL);
+
+  return self;
+}

ext/semian/circuit_breaker.h

@@ -0,0 +1,12 @@
+#ifndef CIRCUIT_BREAKER_H
+#define CIRCUIT_BREAKER_H
+
+#include <ruby.h>
+
+void Init_CircuitBreaker();
+
+VALUE semian_circuit_breaker_alloc(VALUE klass);
+VALUE semian_circuit_breaker_initialize(VALUE self, VALUE id);
+VALUE semian_circuit_breaker_successes(VALUE self);
+
+#endif // CIRCUIT_BREAKER_H

ext/semian/extconf.rb

@@ -23,7 +23,7 @@
 have_func 'rb_thread_blocking_region'
 have_func 'rb_thread_call_without_gvl'
 
-$CFLAGS = "-D_GNU_SOURCE -Werror -Wall "
+$CFLAGS = "-D_GNU_SOURCE -Werror -Wall -std=gnu99 "
 if ENV.key?('DEBUG')
   $CFLAGS << "-O0 -g -DDEBUG"
 else

ext/semian/resource.c

@@ -1,5 +1,7 @@
 #include "resource.h"
 
+#include "util.h"
+
 static VALUE
 cleanup_semian_resource_acquire(VALUE self);
 
@@ -15,9 +17,6 @@ check_tickets_arg(VALUE tickets);
 static long
 check_permissions_arg(VALUE permissions);
 
-static const
-char *check_id_arg(VALUE id);
-
 static double
 check_default_timeout_arg(VALUE default_timeout);
 
@@ -202,22 +201,16 @@ semian_resource_key(VALUE self)
 VALUE
 semian_resource_initialize(VALUE self, VALUE id, VALUE tickets, VALUE quota, VALUE permissions, VALUE default_timeout)
 {
-  long c_permissions;
-  double c_timeout;
-  double c_quota;
-  int c_tickets;
-  semian_resource_t *res = NULL;
-  const char *c_id_str = NULL;
-
   // Check and cast arguments
   check_tickets_xor_quota_arg(tickets, quota);
-  c_quota = check_quota_arg(quota);
-  c_tickets = check_tickets_arg(tickets);
-  c_permissions = check_permissions_arg(permissions);
-  c_id_str = check_id_arg(id);
-  c_timeout = check_default_timeout_arg(default_timeout);
+  double c_quota = check_quota_arg(quota);
+  int c_tickets = check_tickets_arg(tickets);
+  long c_permissions = check_permissions_arg(permissions);
+  const char *c_id_str = check_id_arg(id);
+  double c_timeout = check_default_timeout_arg(default_timeout);
 
   // Build semian resource structure
+  semian_resource_t *res = NULL;
   TypedData_Get_Struct(self, semian_resource_t, &semian_resource_type, res);
 
   // Populate struct fields
@@ -314,23 +307,6 @@ check_tickets_arg(VALUE tickets)
   return c_tickets;
 }
 
-static const char*
-check_id_arg(VALUE id)
-{
-  const char *c_id_str = NULL;
-
-  if (TYPE(id) != T_SYMBOL && TYPE(id) != T_STRING) {
-    rb_raise(rb_eTypeError, "id must be a symbol or string");
-  }
-  if (TYPE(id) == T_SYMBOL) {
-    c_id_str = rb_id2name(rb_to_id(id));
-  } else if (TYPE(id) == T_STRING) {
-    c_id_str = RSTRING_PTR(id);
-  }
-
-  return c_id_str;
-}
-
 static double
 check_default_timeout_arg(VALUE default_timeout)
 {
@@ -361,6 +337,8 @@ static inline void
 semian_resource_free(void *ptr)
 {
   semian_resource_t *res = (semian_resource_t *) ptr;
+  dprintf("Freeing resource sem_id:%d", res->sem_id);
+
   if (res->name) {
     free(res->name);
     res->name = NULL;

ext/semian/semian.c

@@ -1,5 +1,7 @@
 #include "semian.h"
 
+static int use_c_circuits();
+
 void Init_semian()
 {
   VALUE cSemian, cResource;
@@ -64,4 +66,30 @@ void Init_semian()
 
   /* Maximum number of tickets available on this system. */
   rb_define_const(cSemian, "MAX_TICKETS", INT2FIX(system_max_semaphore_count));
+
+  if (use_c_circuits()) {
+    Init_SimpleInteger();
+    Init_SlidingWindow();
+    Init_CircuitBreaker();
+  }
+}
+
+static int
+use_c_circuits() {
+  char *circuit_impl = getenv("SEMIAN_CIRCUIT_BREAKER_IMPL");
+  if (circuit_impl == NULL) {
+    fprintf(stderr, "Warning: Defaulting to worker-based circuit breaker implementation\n");
+    return 0;
+  } else {
+    if (!strcmp(circuit_impl, "worker")) {
+      return 0;
+    } else if (!strcmp(circuit_impl, "host")) {
+      return 1;
+    } else {
+      fprintf(stderr, "Warning: Unknown circuit breaker implementation: '%s'\n", circuit_impl);
+      return 0;
+    }
+  }
+
+  rb_raise(rb_eArgError, "Unknown circuit breaker implementation");
 }

ext/semian/semian.h

@@ -7,7 +7,10 @@ Implements Init_semian, which is used as C/Ruby entrypoint.
 #ifndef SEMIAN_H
 #define SEMIAN_H
 
+#include "circuit_breaker.h"
 #include "resource.h"
+#include "simple_integer.h"
+#include "sliding_window.h"
 
 void Init_semian();