diff options
47 files changed, 863 insertions, 153 deletions
diff --git a/.olddoc.yml b/.olddoc.yml index e005193..e4e8871 100644 --- a/.olddoc.yml +++ b/.olddoc.yml @@ -5,8 +5,11 @@ public_email: raindrops-public@yhbt.net ml_url: - https://yhbt.net/raindrops-public/ - http://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/raindrops-public +imap_url: +- imaps://yhbt.net/inbox.comp.lang.ruby.raindrops.0 +- imap://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/inbox.comp.lang.ruby.raindrops.0 nntp_url: -- nntp://news.public-inbox.org/inbox.comp.lang.ruby.raindrops +- nntps://news.public-inbox.org/inbox.comp.lang.ruby.raindrops - nntp://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/inbox.comp.lang.ruby.raindrops source_code: - git clone https://yhbt.net/raindrops.git diff --git a/GIT-VERSION-GEN b/GIT-VERSION-GEN index 4f8b952..3ee0b87 100755 --- a/GIT-VERSION-GEN +++ b/GIT-VERSION-GEN @@ -1,7 +1,7 @@ #!/bin/sh GVF=GIT-VERSION-FILE -DEF_VER=v0.19.1 +DEF_VER=v0.20.1 LF=' ' @@ -58,24 +58,20 @@ If you use RubyGems: See Raindrops::Middleware and Raindrops::LastDataRecv documentation for use Rack servers. The entire library is fully-documented and we are -responsive on the publically archived mailing list -(mailto:raindrops-public@yhbt.net) if -you have any questions or comments. +responsive on the publicly archived mailbox +(mailto:raindrops-public@yhbt.net) if you have any questions or comments. == Development You can get the latest source via git from the following locations: - git://yhbt.net/raindrops.git - git://repo.or.cz/raindrops.git (mirror) + https://yhbt.net/raindrops.git + http://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/raindrops.git + http://repo.or.cz/w/raindrops.git (gitweb mirror) -You may browse the code from the web and download the latest snapshot -tarballs here: +Snapshots and tarballs are available. -* https://yhbt.net/raindrops.git -* http://repo.or.cz/w/raindrops.git (gitweb) - -Inline patches (from "git format-patch") to the mailing list are +Inline patches (from "git format-patch") to the mailbox are preferred because they allow code review and comments in the reply to the patch. @@ -89,14 +85,27 @@ raindrops is licensed under the LGPL-2.1+ == Contact All feedback (bug reports, user/development discussion, patches, pull -requests) go to the publically archived mailing list: +requests) go to the publicly archived mailbox: mailto:raindrops-public@yhbt.net -Mailing list archives are available over HTTPS and NNTP: +Mail archives are available over HTTP(S), IMAP(S) and NNTP(S): * https://yhbt.net/raindrops-public/ * http://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/raindrops-public/ -* nntp://news.public-inbox.org/inbox.comp.lang.ruby.raindrops +* imaps://yhbt.net/inbox.comp.lang.ruby.raindrops.0 +* imap://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/inbox.comp.lang.ruby.raindrops.0 +* nntps://news.public-inbox.org/inbox.comp.lang.ruby.raindrops +* nntp://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/inbox.comp.lang.ruby.raindrops Since archives are public, scrub sensitive information and use anonymity tools such as Tor or Mixmaster if you deem necessary. + +There is NO WARRANTY whatsoever if anything goes wrong and +no commercial support will ever be provided by the amateur maintainer. +raindrops hackers are NOT responsible for your supply chain security: +read and understand it yourself or get someone you trust to audit it. +Malicious commits and releases will be made if under duress. The only +defense you'll ever have is from reviewing the source code. + +No user or contributor will ever be expected to sacrifice their own +security by running JavaScript or revealing any personal information. diff --git a/examples/linux-listener-stats.rb b/examples/linux-listener-stats.rb index 7e767da..5f67633 100755 --- a/examples/linux-listener-stats.rb +++ b/examples/linux-listener-stats.rb @@ -1,5 +1,6 @@ #!/usr/bin/ruby # -*- encoding: binary -*- +# frozen_string_literal: false $stdout.sync = $stderr.sync = true # this is used to show or watch the number of active and queued # connections on any listener socket from the command line diff --git a/examples/middleware.ru b/examples/middleware.ru index 642016b..a485592 100644 --- a/examples/middleware.ru +++ b/examples/middleware.ru @@ -1,3 +1,4 @@ +# frozen_string_literal: false # sample stand-alone rackup application for Raindrops::Middleware require 'rack/lobster' require 'raindrops' diff --git a/examples/watcher.ru b/examples/watcher.ru index a3e7fdb..e2aa97c 100644 --- a/examples/watcher.ru +++ b/examples/watcher.ru @@ -1,3 +1,4 @@ +# frozen_string_literal: false # Sample standalone Rack application, recommended use is with Zbatery # See zbatery.conf.rb require "raindrops" diff --git a/examples/watcher_demo.ru b/examples/watcher_demo.ru index 91f4cca..7a6e675 100644 --- a/examples/watcher_demo.ru +++ b/examples/watcher_demo.ru @@ -1,3 +1,4 @@ +# frozen_string_literal: false # This is a snippet of the config that powers # https://yhbt.net/raindrops-demo/ # This may be used with the packaged zbatery.conf.rb diff --git a/examples/yahns.conf.rb b/examples/yahns.conf.rb index f5b4f10..75f0bd1 100644 --- a/examples/yahns.conf.rb +++ b/examples/yahns.conf.rb @@ -1,3 +1,4 @@ +# frozen_string_literal: false # Inlined rack app using yahns server (git clone git://yhbt.net/yahns.git) # Usage: yahns -c /path/to/this/file.conf.rb # There is no separate config.ru file for this example, diff --git a/examples/zbatery.conf.rb b/examples/zbatery.conf.rb index 5f94c0e..0537466 100644 --- a/examples/zbatery.conf.rb +++ b/examples/zbatery.conf.rb @@ -1,3 +1,4 @@ +# frozen_string_literal: false # Used for running Raindrops::Watcher, which requires a multi-threaded # Rack server capable of streaming a response. Threads must be used, # so any multi-threaded Rack server may be used. diff --git a/ext/raindrops/extconf.rb b/ext/raindrops/extconf.rb index 792e509..b1310b0 100644 --- a/ext/raindrops/extconf.rb +++ b/ext/raindrops/extconf.rb @@ -1,9 +1,12 @@ +# frozen_string_literal: false require 'mkmf' require 'shellwords' +$CFLAGS += ' -O0 ' # faster checks dir_config('atomic_ops') have_func('mmap', 'sys/mman.h') or abort 'mmap() not found' have_func('munmap', 'sys/mman.h') or abort 'munmap() not found' +have_func('rb_io_descriptor') $CPPFLAGS += " -D_GNU_SOURCE " have_func('mremap', 'sys/mman.h') @@ -157,4 +160,5 @@ Users of Debian-based distros may run: apt-get install libatomic-ops-dev SRC create_header # generate extconf.h to avoid excessively long command-line +$CFLAGS.sub!(/ -O0 /, '') create_makefile('raindrops_ext') diff --git a/ext/raindrops/khashl.h b/ext/raindrops/khashl.h new file mode 100644 index 0000000..9a6e4fe --- /dev/null +++ b/ext/raindrops/khashl.h @@ -0,0 +1,444 @@ +/* The MIT License + + Copyright (c) 2019-2023 by Attractive Chaos <attractor@live.co.uk> + + Permission is hereby granted, free of charge, to any person obtaining + a copy of this software and associated documentation files (the + "Software"), to deal in the Software without restriction, including + without limitation the rights to use, copy, modify, merge, publish, + distribute, sublicense, and/or sell copies of the Software, and to + permit persons to whom the Software is furnished to do so, subject to + the following conditions: + + The above copyright notice and this permission notice shall be + included in all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF + MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS + BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN + ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE. +*/ + +#ifndef __AC_KHASHL_H +#define __AC_KHASHL_H + +#define AC_VERSION_KHASHL_H "0.2" + +#include <stdlib.h> +#include <string.h> +#include <limits.h> + +/************************************ + * Compiler specific configurations * + ************************************/ + +#if UINT_MAX == 0xffffffffu +typedef unsigned int khint32_t; +#elif ULONG_MAX == 0xffffffffu +typedef unsigned long khint32_t; +#endif + +#if ULONG_MAX == ULLONG_MAX +typedef unsigned long khint64_t; +#else +typedef unsigned long long khint64_t; +#endif + +#ifndef kh_inline +#ifdef _MSC_VER +#define kh_inline __inline +#else +#define kh_inline inline +#endif +#endif /* kh_inline */ + +#ifndef klib_unused +#if (defined __clang__ && __clang_major__ >= 3) || (defined __GNUC__ && __GNUC__ >= 3) +#define klib_unused __attribute__ ((__unused__)) +#else +#define klib_unused +#endif +#endif /* klib_unused */ + +#define KH_LOCAL static kh_inline klib_unused + +typedef khint32_t khint_t; + +/****************** + * malloc aliases * + ******************/ + +#ifndef kcalloc +#define kcalloc(N,Z) calloc(N,Z) +#endif +#ifndef kmalloc +#define kmalloc(Z) malloc(Z) +#endif +#ifndef krealloc +#define krealloc(P,Z) realloc(P,Z) +#endif +#ifndef kfree +#define kfree(P) free(P) +#endif + +/**************************** + * Simple private functions * + ****************************/ + +#define __kh_used(flag, i) (flag[i>>5] >> (i&0x1fU) & 1U) +#define __kh_set_used(flag, i) (flag[i>>5] |= 1U<<(i&0x1fU)) +#define __kh_set_unused(flag, i) (flag[i>>5] &= ~(1U<<(i&0x1fU))) + +#define __kh_fsize(m) ((m) < 32? 1 : (m)>>5) + +static kh_inline khint_t __kh_h2b(khint_t hash, khint_t bits) { return hash * 2654435769U >> (32 - bits); } + +/******************* + * Hash table base * + *******************/ + +#define __KHASHL_TYPE(HType, khkey_t) \ + typedef struct HType { \ + khint_t bits, count; \ + khint32_t *used; \ + khkey_t *keys; \ + } HType; + +#define __KHASHL_PROTOTYPES(HType, prefix, khkey_t) \ + extern HType *prefix##_init(void); \ + extern void prefix##_destroy(HType *h); \ + extern void prefix##_clear(HType *h); \ + extern khint_t prefix##_getp(const HType *h, const khkey_t *key); \ + extern int prefix##_resize(HType *h, khint_t new_n_buckets); \ + extern khint_t prefix##_putp(HType *h, const khkey_t *key, int *absent); \ + extern void prefix##_del(HType *h, khint_t k); + +#define __KHASHL_IMPL_BASIC(SCOPE, HType, prefix) \ + SCOPE HType *prefix##_init(void) { \ + return (HType*)kcalloc(1, sizeof(HType)); \ + } \ + SCOPE void prefix##_destroy(HType *h) { \ + if (!h) return; \ + kfree((void *)h->keys); kfree(h->used); \ + kfree(h); \ + } \ + SCOPE void prefix##_clear(HType *h) { \ + if (h && h->used) { \ + khint_t n_buckets = (khint_t)1U << h->bits; \ + memset(h->used, 0, __kh_fsize(n_buckets) * sizeof(khint32_t)); \ + h->count = 0; \ + } \ + } + +#define __KHASHL_IMPL_GET(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + SCOPE khint_t prefix##_getp_core(const HType *h, const khkey_t *key, khint_t hash) { \ + khint_t i, last, n_buckets, mask; \ + if (h->keys == 0) return 0; \ + n_buckets = (khint_t)1U << h->bits; \ + mask = n_buckets - 1U; \ + i = last = __kh_h2b(hash, h->bits); \ + while (__kh_used(h->used, i) && !__hash_eq(h->keys[i], *key)) { \ + i = (i + 1U) & mask; \ + if (i == last) return n_buckets; \ + } \ + return !__kh_used(h->used, i)? n_buckets : i; \ + } \ + SCOPE khint_t prefix##_getp(const HType *h, const khkey_t *key) { return prefix##_getp_core(h, key, __hash_fn(*key)); } \ + SCOPE khint_t prefix##_get(const HType *h, khkey_t key) { return prefix##_getp_core(h, &key, __hash_fn(key)); } + +#define __KHASHL_IMPL_RESIZE(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + SCOPE int prefix##_resize(HType *h, khint_t new_n_buckets) { \ + khint32_t *new_used = 0; \ + khint_t j = 0, x = new_n_buckets, n_buckets, new_bits, new_mask; \ + while ((x >>= 1) != 0) ++j; \ + if (new_n_buckets & (new_n_buckets - 1)) ++j; \ + new_bits = j > 2? j : 2; \ + new_n_buckets = (khint_t)1U << new_bits; \ + if (h->count > (new_n_buckets>>1) + (new_n_buckets>>2)) return 0; /* requested size is too small */ \ + new_used = (khint32_t*)kcalloc(__kh_fsize(new_n_buckets), \ + sizeof(khint32_t)); \ + if (!new_used) return -1; /* not enough memory */ \ + n_buckets = h->keys? (khint_t)1U<<h->bits : 0U; \ + if (n_buckets < new_n_buckets) { /* expand */ \ + h->keys = ruby_xrealloc2(h->keys, new_n_buckets, \ + sizeof(khkey_t)); \ + } /* otherwise shrink */ \ + new_mask = new_n_buckets - 1; \ + for (j = 0; j != n_buckets; ++j) { \ + khkey_t key; \ + if (!__kh_used(h->used, j)) continue; \ + key = h->keys[j]; \ + __kh_set_unused(h->used, j); \ + while (1) { /* kick-out process; sort of like in Cuckoo hashing */ \ + khint_t i; \ + i = __kh_h2b(__hash_fn(key), new_bits); \ + while (__kh_used(new_used, i)) i = (i + 1) & new_mask; \ + __kh_set_used(new_used, i); \ + if (i < n_buckets && __kh_used(h->used, i)) { /* kick out the existing element */ \ + { khkey_t tmp = h->keys[i]; h->keys[i] = key; key = tmp; } \ + __kh_set_unused(h->used, i); /* mark it as deleted in the old hash table */ \ + } else { /* write the element and jump out of the loop */ \ + h->keys[i] = key; \ + break; \ + } \ + } \ + } \ + if (n_buckets > new_n_buckets) /* shrink the hash table */ \ + h->keys = ruby_xrealloc2(h->keys, new_n_buckets, \ + sizeof(khkey_t)); \ + kfree(h->used); /* free the working space */ \ + h->used = new_used, h->bits = new_bits; \ + return 0; \ + } + +#define __KHASHL_IMPL_PUT(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + SCOPE khint_t prefix##_putp_core(HType *h, const khkey_t *key, khint_t hash, int *absent) { \ + khint_t n_buckets, i, last, mask; \ + n_buckets = h->keys? (khint_t)1U<<h->bits : 0U; \ + *absent = -1; \ + if (h->count >= (n_buckets>>1) + (n_buckets>>2)) { /* rehashing */ \ + if (prefix##_resize(h, n_buckets + 1U) < 0) \ + return n_buckets; \ + n_buckets = (khint_t)1U<<h->bits; \ + } /* TODO: to implement automatically shrinking; resize() already support shrinking */ \ + mask = n_buckets - 1; \ + i = last = __kh_h2b(hash, h->bits); \ + while (__kh_used(h->used, i) && !__hash_eq(h->keys[i], *key)) { \ + i = (i + 1U) & mask; \ + if (i == last) break; \ + } \ + if (!__kh_used(h->used, i)) { /* not present at all */ \ + h->keys[i] = *key; \ + __kh_set_used(h->used, i); \ + ++h->count; \ + *absent = 1; \ + } else *absent = 0; /* Don't touch h->keys[i] if present */ \ + return i; \ + } \ + SCOPE khint_t prefix##_putp(HType *h, const khkey_t *key, int *absent) { return prefix##_putp_core(h, key, __hash_fn(*key), absent); } \ + SCOPE khint_t prefix##_put(HType *h, khkey_t key, int *absent) { return prefix##_putp_core(h, &key, __hash_fn(key), absent); } + +#define __KHASHL_IMPL_DEL(SCOPE, HType, prefix, khkey_t, __hash_fn) \ + SCOPE int prefix##_del(HType *h, khint_t i) { \ + khint_t j = i, k, mask, n_buckets; \ + if (h->keys == 0) return 0; \ + n_buckets = (khint_t)1U<<h->bits; \ + mask = n_buckets - 1U; \ + while (1) { \ + j = (j + 1U) & mask; \ + if (j == i || !__kh_used(h->used, j)) break; /* j==i only when the table is completely full */ \ + k = __kh_h2b(__hash_fn(h->keys[j]), h->bits); \ + if ((j > i && (k <= i || k > j)) || (j < i && (k <= i && k > j))) \ + h->keys[i] = h->keys[j], i = j; \ + } \ + __kh_set_unused(h->used, i); \ + --h->count; \ + return 1; \ + } + +#define KHASHL_DECLARE(HType, prefix, khkey_t) \ + __KHASHL_TYPE(HType, khkey_t) \ + __KHASHL_PROTOTYPES(HType, prefix, khkey_t) + +#define KHASHL_INIT(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + __KHASHL_TYPE(HType, khkey_t) \ + __KHASHL_IMPL_BASIC(SCOPE, HType, prefix) \ + __KHASHL_IMPL_GET(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + __KHASHL_IMPL_RESIZE(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + __KHASHL_IMPL_PUT(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + __KHASHL_IMPL_DEL(SCOPE, HType, prefix, khkey_t, __hash_fn) + +/*************************** + * Ensemble of hash tables * + ***************************/ + +typedef struct { + khint_t sub, pos; +} kh_ensitr_t; + +#define KHASHE_INIT(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + KHASHL_INIT(KH_LOCAL, HType##_sub, prefix##_sub, khkey_t, __hash_fn, __hash_eq) \ + typedef struct HType { \ + khint64_t count:54, bits:8; \ + HType##_sub *sub; \ + } HType; \ + SCOPE HType *prefix##_init(int bits) { \ + HType *g; \ + g = (HType*)kcalloc(1, sizeof(*g)); \ + g->bits = bits; \ + g->sub = (HType##_sub*)kcalloc(1U<<bits, sizeof(*g->sub)); \ + return g; \ + } \ + SCOPE void prefix##_destroy(HType *g) { \ + int t; \ + if (!g) return; \ + for (t = 0; t < 1<<g->bits; ++t) { kfree((void*)g->sub[t].keys); kfree(g->sub[t].used); } \ + kfree(g->sub); kfree(g); \ + } \ + SCOPE kh_ensitr_t prefix##_getp(const HType *g, const khkey_t *key) { \ + khint_t hash, low, ret; \ + kh_ensitr_t r; \ + HType##_sub *h; \ + hash = __hash_fn(*key); \ + low = hash & ((1U<<g->bits) - 1); \ + h = &g->sub[low]; \ + ret = prefix##_sub_getp_core(h, key, hash); \ + if (ret == 1U<<h->bits) r.sub = low, r.pos = (khint_t)-1; \ + else r.sub = low, r.pos = ret; \ + return r; \ + } \ + SCOPE kh_ensitr_t prefix##_get(const HType *g, const khkey_t key) { return prefix##_getp(g, &key); } \ + SCOPE kh_ensitr_t prefix##_putp(HType *g, const khkey_t *key, int *absent) { \ + khint_t hash, low, ret; \ + kh_ensitr_t r; \ + HType##_sub *h; \ + hash = __hash_fn(*key); \ + low = hash & ((1U<<g->bits) - 1); \ + h = &g->sub[low]; \ + ret = prefix##_sub_putp_core(h, key, hash, absent); \ + if (*absent) ++g->count; \ + if (ret == 1U<<h->bits) r.sub = low, r.pos = (khint_t)-1; \ + else r.sub = low, r.pos = ret; \ + return r; \ + } \ + SCOPE kh_ensitr_t prefix##_put(HType *g, const khkey_t key, int *absent) { return prefix##_putp(g, &key, absent); } \ + SCOPE int prefix##_del(HType *g, kh_ensitr_t itr) { \ + HType##_sub *h = &g->sub[itr.sub]; \ + int ret; \ + ret = prefix##_sub_del(h, itr.pos); \ + if (ret) --g->count; \ + return ret; \ + } + +/***************************** + * More convenient interface * + *****************************/ + +#define __kh_packed __attribute__ ((__packed__)) +#define __kh_cached_hash(x) ((x).hash) + +#define KHASHL_SET_INIT(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + typedef struct { khkey_t key; } __kh_packed HType##_s_bucket_t; \ + static kh_inline khint_t prefix##_s_hash(HType##_s_bucket_t x) { return __hash_fn(x.key); } \ + static kh_inline int prefix##_s_eq(HType##_s_bucket_t x, HType##_s_bucket_t y) { return __hash_eq(x.key, y.key); } \ + KHASHL_INIT(KH_LOCAL, HType, prefix##_s, HType##_s_bucket_t, prefix##_s_hash, prefix##_s_eq) \ + SCOPE HType *prefix##_init(void) { return prefix##_s_init(); } \ + SCOPE void prefix##_destroy(HType *h) { prefix##_s_destroy(h); } \ + SCOPE void prefix##_resize(HType *h, khint_t new_n_buckets) { prefix##_s_resize(h, new_n_buckets); } \ + SCOPE khint_t prefix##_get(const HType *h, khkey_t key) { HType##_s_bucket_t t; t.key = key; return prefix##_s_getp(h, &t); } \ + SCOPE int prefix##_del(HType *h, khint_t k) { return prefix##_s_del(h, k); } \ + SCOPE khint_t prefix##_put(HType *h, khkey_t key, int *absent) { HType##_s_bucket_t t; t.key = key; return prefix##_s_putp(h, &t, absent); } + +#define KHASHL_MAP_INIT(SCOPE, HType, prefix, khkey_t, kh_val_t, __hash_fn, __hash_eq) \ + typedef struct { khkey_t key; kh_val_t val; } __kh_packed HType##_m_bucket_t; \ + static kh_inline khint_t prefix##_m_hash(HType##_m_bucket_t x) { return __hash_fn(x.key); } \ + static kh_inline int prefix##_m_eq(HType##_m_bucket_t x, HType##_m_bucket_t y) { return __hash_eq(x.key, y.key); } \ + KHASHL_INIT(KH_LOCAL, HType, prefix##_m, HType##_m_bucket_t, prefix##_m_hash, prefix##_m_eq) \ + SCOPE HType *prefix##_init(void) { return prefix##_m_init(); } \ + SCOPE void prefix##_destroy(HType *h) { prefix##_m_destroy(h); } \ + SCOPE khint_t prefix##_get(const HType *h, khkey_t key) { HType##_m_bucket_t t; t.key = key; return prefix##_m_getp(h, &t); } \ + SCOPE int prefix##_del(HType *h, khint_t k) { return prefix##_m_del(h, k); } \ + SCOPE khint_t prefix##_put(HType *h, khkey_t key, int *absent) { HType##_m_bucket_t t; t.key = key; return prefix##_m_putp(h, &t, absent); } + +#define KHASHL_CSET_INIT(SCOPE, HType, prefix, khkey_t, __hash_fn, __hash_eq) \ + typedef struct { khkey_t key; khint_t hash; } __kh_packed HType##_cs_bucket_t; \ + static kh_inline int prefix##_cs_eq(HType##_cs_bucket_t x, HType##_cs_bucket_t y) { return x.hash == y.hash && __hash_eq(x.key, y.key); } \ + KHASHL_INIT(KH_LOCAL, HType, prefix##_cs, HType##_cs_bucket_t, __kh_cached_hash, prefix##_cs_eq) \ + SCOPE HType *prefix##_init(void) { return prefix##_cs_init(); } \ + SCOPE void prefix##_destroy(HType *h) { prefix##_cs_destroy(h); } \ + SCOPE khint_t prefix##_get(const HType *h, khkey_t key) { HType##_cs_bucket_t t; t.key = key; t.hash = __hash_fn(key); return prefix##_cs_getp(h, &t); } \ + SCOPE int prefix##_del(HType *h, khint_t k) { return prefix##_cs_del(h, k); } \ + SCOPE khint_t prefix##_put(HType *h, khkey_t key, int *absent) { HType##_cs_bucket_t t; t.key = key, t.hash = __hash_fn(key); return prefix##_cs_putp(h, &t, absent); } + +#define KHASHL_CMAP_INIT(SCOPE, HType, prefix, khkey_t, kh_val_t, __hash_fn, __hash_eq) \ + typedef struct { khkey_t key; kh_val_t val; khint_t hash; } __kh_packed HType##_cm_bucket_t; \ + static kh_inline int prefix##_cm_eq(HType##_cm_bucket_t x, HType##_cm_bucket_t y) { return x.hash == y.hash && __hash_eq(x.key, y.key); } \ + KHASHL_INIT(KH_LOCAL, HType, prefix##_cm, HType##_cm_bucket_t, __kh_cached_hash, prefix##_cm_eq) \ + SCOPE HType *prefix##_init(void) { return prefix##_cm_init(); } \ + SCOPE void prefix##_destroy(HType *h) { prefix##_cm_destroy(h); } \ + SCOPE khint_t prefix##_get(const HType *h, khkey_t key) { HType##_cm_bucket_t t; t.key = key; t.hash = __hash_fn(key); return prefix##_cm_getp(h, &t); } \ + SCOPE int prefix##_del(HType *h, khint_t k) { return prefix##_cm_del(h, k); } \ + SCOPE khint_t prefix##_put(HType *h, khkey_t key, int *absent) { HType##_cm_bucket_t t; t.key = key, t.hash = __hash_fn(key); return prefix##_cm_putp(h, &t, absent); } + +#define KHASHE_MAP_INIT(SCOPE, HType, prefix, khkey_t, kh_val_t, __hash_fn, __hash_eq) \ + typedef struct { khkey_t key; kh_val_t val; } __kh_packed HType##_m_bucket_t; \ + static kh_inline khint_t prefix##_m_hash(HType##_m_bucket_t x) { return __hash_fn(x.key); } \ + static kh_inline int prefix##_m_eq(HType##_m_bucket_t x, HType##_m_bucket_t y) { return __hash_eq(x.key, y.key); } \ + KHASHE_INIT(KH_LOCAL, HType, prefix##_m, HType##_m_bucket_t, prefix##_m_hash, prefix##_m_eq) \ + SCOPE HType *prefix##_init(int bits) { return prefix##_m_init(bits); } \ + SCOPE void prefix##_destroy(HType *h) { prefix##_m_destroy(h); } \ + SCOPE kh_ensitr_t prefix##_get(const HType *h, khkey_t key) { HType##_m_bucket_t t; t.key = key; return prefix##_m_getp(h, &t); } \ + SCOPE int prefix##_del(HType *h, kh_ensitr_t k) { return prefix##_m_del(h, k); } \ + SCOPE kh_ensitr_t prefix##_put(HType *h, khkey_t key, int *absent) { HType##_m_bucket_t t; t.key = key; return prefix##_m_putp(h, &t, absent); } + +/************************** + * Public macro functions * + **************************/ + +#define kh_bucket(h, x) ((h)->keys[x]) +#define kh_size(h) ((h)->count) +#define kh_capacity(h) ((h)->keys? 1U<<(h)->bits : 0U) +#define kh_end(h) kh_capacity(h) + +#define kh_key(h, x) ((h)->keys[x].key) +#define kh_val(h, x) ((h)->keys[x].val) +#define kh_exist(h, x) __kh_used((h)->used, (x)) + +#define kh_ens_key(g, x) kh_key(&(g)->sub[(x).sub], (x).pos) +#define kh_ens_val(g, x) kh_val(&(g)->sub[(x).sub], (x).pos) +#define kh_ens_exist(g, x) kh_exist(&(g)->sub[(x).sub], (x).pos) +#define kh_ens_is_end(x) ((x).pos == (khint_t)-1) +#define kh_ens_size(g) ((g)->count) + +/************************************** + * Common hash and equality functions * + **************************************/ + +#define kh_eq_generic(a, b) ((a) == (b)) +#define kh_eq_str(a, b) (strcmp((a), (b)) == 0) +#define kh_hash_dummy(x) ((khint_t)(x)) + +static kh_inline khint_t kh_hash_uint32(khint_t key) { + key += ~(key << 15); + key ^= (key >> 10); + key += (key << 3); + key ^= (key >> 6); + key += ~(key << 11); + key ^= (key >> 16); + return key; +} + +static kh_inline khint_t kh_hash_uint64(khint64_t key) { + key = ~key + (key << 21); + key = key ^ key >> 24; + key = (key + (key << 3)) + (key << 8); + key = key ^ key >> 14; + key = (key + (key << 2)) + (key << 4); + key = key ^ key >> 28; + key = key + (key << 31); + return (khint_t)key; +} + +#define KH_FNV_SEED 11 + +static kh_inline khint_t kh_hash_str(const char *s) { /* FNV1a */ + khint_t h = KH_FNV_SEED ^ 2166136261U; + const unsigned char *t = (const unsigned char*)s; + for (; *t; ++t) + h ^= *t, h *= 16777619; + return h; +} + +static kh_inline khint_t kh_hash_bytes(int len, const unsigned char *s) { + khint_t h = KH_FNV_SEED ^ 2166136261U; + int i; + for (i = 0; i < len; ++i) + h ^= s[i], h *= 16777619; + return h; +} + +#endif /* __AC_KHASHL_H */ diff --git a/ext/raindrops/linux_inet_diag.c b/ext/raindrops/linux_inet_diag.c index cabd427..79f24bb 100644 --- a/ext/raindrops/linux_inet_diag.c +++ b/ext/raindrops/linux_inet_diag.c @@ -1,6 +1,5 @@ #include <ruby.h> #include <stdarg.h> -#include <ruby/st.h> #include "my_fileno.h" #ifdef __linux__ @@ -54,12 +53,23 @@ struct listen_stats { uint32_t listener_p; }; +/* override khashl.h defaults, these run w/o GVL */ +#define kcalloc(N,Z) xcalloc(N,Z) +#define kmalloc(Z) xmalloc(Z) +#define krealloc(P,Z) abort() /* never called, we use ruby_xrealloc2 */ +#define kfree(P) xfree(P) + +#include "khashl.h" +KHASHL_CMAP_INIT(KH_LOCAL, addr2stats /* type */, a2s /* pfx */, + char * /* key */, struct listen_stats * /* val */, + kh_hash_str, kh_eq_str) + #define OPLEN (sizeof(struct inet_diag_bc_op) + \ sizeof(struct inet_diag_hostcond) + \ sizeof(struct sockaddr_storage)) struct nogvl_args { - st_table *table; + addr2stats *a2s; struct iovec iov[3]; /* last iov holds inet_diag bytecode */ struct listen_stats stats; int fd; @@ -106,14 +116,6 @@ static VALUE rb_listen_stats(struct listen_stats *stats) return rb_struct_new(cListenStats, active, queued); } -static int st_free_data(st_data_t key, st_data_t value, st_data_t ignored) -{ - xfree((void *)key); - xfree((void *)value); - - return ST_DELETE; -} - /* * call-seq: * remove_scope_id(ip_address) @@ -151,36 +153,6 @@ static VALUE remove_scope_id(const char *addr) return rv; } -static int st_to_hash(st_data_t key, st_data_t value, VALUE hash) -{ - struct listen_stats *stats = (struct listen_stats *)value; - - if (stats->listener_p) { - VALUE k = remove_scope_id((const char *)key); - VALUE v = rb_listen_stats(stats); - - OBJ_FREEZE(k); - rb_hash_aset(hash, k, v); - } - return st_free_data(key, value, 0); -} - -static int st_AND_hash(st_data_t key, st_data_t value, VALUE hash) -{ - struct listen_stats *stats = (struct listen_stats *)value; - - if (stats->listener_p) { - VALUE k = remove_scope_id((const char *)key); - - if (rb_hash_lookup(hash, k) == Qtrue) { - VALUE v = rb_listen_stats(stats); - OBJ_FREEZE(k); - rb_hash_aset(hash, k, v); - } - } - return st_free_data(key, value, 0); -} - static const char *addr_any(sa_family_t family) { static const char ipv4[] = "0.0.0.0"; @@ -209,32 +181,36 @@ static void bug_warn_nogvl(const char *fmt, ...) fflush(stderr); } -static struct listen_stats *stats_for(st_table *table, struct inet_diag_msg *r) +static struct listen_stats *stats_for(addr2stats *a2s, struct inet_diag_msg *r) { char *host, *key, *port, *old_key; - size_t alloca_len; struct listen_stats *stats; socklen_t hostlen; socklen_t portlen = (socklen_t)sizeof("65535"); - int n; + int n, absent; const void *src = r->id.idiag_src; + char buf[INET6_ADDRSTRLEN]; + size_t buf_len; + khint_t ki; switch (r->idiag_family) { case AF_INET: { hostlen = INET_ADDRSTRLEN; - alloca_len = hostlen + portlen; - host = key = alloca(alloca_len); + buf_len = hostlen + portlen; + host = key = buf; break; } case AF_INET6: { hostlen = INET6_ADDRSTRLEN; - alloca_len = 1 + hostlen + 1 + portlen; - key = alloca(alloca_len); + buf_len = 1 + hostlen + 1 + portlen; + key = buf; host = key + 1; break; } default: - assert(0 && "unsupported address family, could that be IPv7?!"); + fprintf(stderr, "unsupported .idiag_family: %u\n", + (unsigned)r->idiag_family); + return NULL; /* can't raise w/o GVL */ } if (!inet_ntop(r->idiag_family, src, host, hostlen)) { bug_warn_nogvl("BUG: inet_ntop: %s\n", strerror(errno)); @@ -254,7 +230,8 @@ static struct listen_stats *stats_for(st_table *table, struct inet_diag_msg *r) port = host + hostlen + 2; break; default: - assert(0 && "unsupported address family, could that be IPv7?!"); + assert(0 && "should never get here (returned above)"); + abort(); } n = snprintf(port, portlen, "%u", ntohs(r->id.idiag_sport)); @@ -263,21 +240,24 @@ static struct listen_stats *stats_for(st_table *table, struct inet_diag_msg *r) *key = '\0'; } - if (st_lookup(table, (st_data_t)key, (st_data_t *)&stats)) - return stats; + ki = a2s_get(a2s, key); + if (ki < kh_end(a2s)) + return kh_val(a2s, ki); old_key = key; if (r->idiag_state == TCP_ESTABLISHED) { - n = snprintf(key, alloca_len, "%s:%u", + n = snprintf(key, buf_len, "%s:%u", addr_any(r->idiag_family), ntohs(r->id.idiag_sport)); if (n <= 0) { bug_warn_nogvl("BUG: snprintf: %d\n", n); *key = '\0'; } - if (st_lookup(table, (st_data_t)key, (st_data_t *)&stats)) - return stats; + + ki = a2s_get(a2s, key); + if (ki < kh_end(a2s)) + return kh_val(a2s, ki); if (n <= 0) { key = xmalloc(1); *key = '\0'; @@ -292,21 +272,25 @@ static struct listen_stats *stats_for(st_table *table, struct inet_diag_msg *r) memcpy(key, old_key, old_len); } stats = xcalloc(1, sizeof(struct listen_stats)); - st_insert(table, (st_data_t)key, (st_data_t)stats); + ki = a2s_put(a2s, key, &absent); /* fails on OOM due to xrealloc */ + assert(absent > 0 && "redundant put"); + kh_val(a2s, ki) = stats; return stats; } -static void table_incr_active(st_table *table, struct inet_diag_msg *r) +static void table_incr_active(addr2stats *a2s, struct inet_diag_msg *r) { - struct listen_stats *stats = stats_for(table, r); + struct listen_stats *stats = stats_for(a2s, r); + if (!stats) return; ++stats->active; } -static void table_set_queued(st_table *table, struct inet_diag_msg *r) +static void table_set_queued(addr2stats *a2s, struct inet_diag_msg *r) { - struct listen_stats *stats = stats_for(table, r); + struct listen_stats *stats = stats_for(a2s, r); + if (!stats) return; stats->listener_p = 1; - stats->queued = r->idiag_rqueue; + stats->queued += r->idiag_rqueue; } /* inner loop of inet_diag, called for every socket returned by netlink */ @@ -320,15 +304,15 @@ static inline void r_acc(struct nogvl_args *args, struct inet_diag_msg *r) if (r->idiag_inode == 0) return; if (r->idiag_state == TCP_ESTABLISHED) { - if (args->table) - table_incr_active(args->table, r); + if (args->a2s) + table_incr_active(args->a2s, r); else args->stats.active++; } else { /* if (r->idiag_state == TCP_LISTEN) */ - if (args->table) - table_set_queued(args->table, r); + if (args->a2s) + table_set_queued(args->a2s, r); else - args->stats.queued = r->idiag_rqueue; + args->stats.queued += r->idiag_rqueue; } /* * we wont get anything else because of the idiag_states filter @@ -444,11 +428,18 @@ static VALUE diag(void *ptr) } out: /* prepare to raise, free memory before reacquiring GVL */ - if (err && args->table) { + if (err && args->a2s) { int save_errno = errno; + khint_t ki; + + /* no kh_foreach* in khashl.h (unlike original khash.h) */ + for (ki = 0; ki < kh_end(args->a2s); ki++) { + if (!kh_exist(args->a2s, ki)) continue; - st_foreach(args->table, st_free_data, 0); - st_free_table(args->table); + xfree(kh_key(args->a2s, ki)); + xfree(kh_val(args->a2s, ki)); + } + a2s_destroy(args->a2s); errno = save_errno; } return (VALUE)err; @@ -564,7 +555,7 @@ static void gen_bytecode(struct iovec *iov, union any_addr *inet) /* * n.b. we may safely raise here because an error will cause diag() - * to free args->table + * to free args->a2s */ static void nl_errcheck(VALUE r) { @@ -591,6 +582,7 @@ static VALUE tcp_stats(struct nogvl_args *args, VALUE addr) return rb_listen_stats(&args->stats); } +/* part of the Ruby rb_hash_* API still relies on st_data_t... */ static int drop_placeholders(st_data_t k, st_data_t v, st_data_t ign) { if ((VALUE)v == Qtrue) @@ -615,7 +607,10 @@ static VALUE tcp_listener_stats(int argc, VALUE *argv, VALUE self) { VALUE rv = rb_hash_new(); struct nogvl_args args; - VALUE addrs, sock; + VALUE addrs, sock, buf; + khint_t ki; + struct listen_stats *stats; + char *key; rb_scan_args(argc, argv, "02", &addrs, &sock); @@ -624,17 +619,18 @@ static VALUE tcp_listener_stats(int argc, VALUE *argv, VALUE self) * buffer for recvmsg() later, we already checked for * OPLEN <= page_size at initialization */ + buf = rb_str_buf_new(page_size); args.iov[2].iov_len = OPLEN; - args.iov[2].iov_base = alloca(page_size); - args.table = NULL; - if (NIL_P(sock)) - sock = rb_funcall(cIDSock, id_new, 0); + args.iov[2].iov_base = RSTRING_PTR(buf); + args.a2s = NULL; + sock = NIL_P(sock) ? rb_funcall(cIDSock, id_new, 0) + : rb_io_get_io(sock); args.fd = my_fileno(sock); switch (TYPE(addrs)) { case T_STRING: rb_hash_aset(rv, addrs, tcp_stats(&args, addrs)); - return rv; + goto out; case T_ARRAY: { long i; long len = RARRAY_LEN(addrs); @@ -643,7 +639,7 @@ static VALUE tcp_listener_stats(int argc, VALUE *argv, VALUE self) VALUE cur = rb_ary_entry(addrs, 0); rb_hash_aset(rv, cur, tcp_stats(&args, cur)); - return rv; + goto out; } for (i = 0; i < len; i++) { union any_addr check; @@ -655,23 +651,38 @@ static VALUE tcp_listener_stats(int argc, VALUE *argv, VALUE self) /* fall through */ } case T_NIL: - args.table = st_init_strtable(); + args.a2s = a2s_init(); gen_bytecode_all(&args.iov[2]); break; default: + if (argc < 2) rb_io_close(sock); rb_raise(rb_eArgError, "addr must be an array of strings, a string, or nil"); } nl_errcheck(rd_fd_region(diag, &args, args.fd)); - st_foreach(args.table, NIL_P(addrs) ? st_to_hash : st_AND_hash, rv); - st_free_table(args.table); + /* no kh_foreach* in khashl.h (unlike original khash.h) */ + for (ki = 0; ki < kh_end(args.a2s); ki++) { + if (!kh_exist(args.a2s, ki)) continue; + key = kh_key(args.a2s, ki); + stats = kh_val(args.a2s, ki); + if (stats->listener_p) { + VALUE k = remove_scope_id(key); + if (NIL_P(addrs) || rb_hash_lookup(rv, k) == Qtrue) + rb_hash_aset(rv, k, rb_listen_stats(stats)); + } + xfree(key); + xfree(stats); + } + a2s_destroy(args.a2s); if (RHASH_SIZE(rv) > 1) rb_hash_foreach(rv, drop_placeholders, Qfalse); +out: /* let GC deal with corner cases */ + rb_str_resize(buf, 0); if (argc < 2) rb_io_close(sock); return rv; } diff --git a/ext/raindrops/my_fileno.h b/ext/raindrops/my_fileno.h index bdf1a5f..3a0100f 100644 --- a/ext/raindrops/my_fileno.h +++ b/ext/raindrops/my_fileno.h @@ -1,36 +1,16 @@ #include <ruby.h> -#ifdef HAVE_RUBY_IO_H -# include <ruby/io.h> -#else -# include <stdio.h> -# include <rubyio.h> -#endif - -#if ! HAVE_RB_IO_T -# define rb_io_t OpenFile -#endif - -#ifdef GetReadFile -# define FPTR_TO_FD(fptr) (fileno(GetReadFile(fptr))) -#else -# if !HAVE_RB_IO_T || (RUBY_VERSION_MAJOR == 1 && RUBY_VERSION_MINOR == 8) -# define FPTR_TO_FD(fptr) fileno(fptr->f) -# else -# define FPTR_TO_FD(fptr) fptr->fd -# endif -#endif +#include <ruby/io.h> +#ifdef HAVE_RB_IO_DESCRIPTOR +# define my_fileno(io) rb_io_descriptor(io) +#else /* Ruby <3.1 */ static int my_fileno(VALUE io) { rb_io_t *fptr; - int fd; - if (TYPE(io) != T_FILE) - io = rb_convert_type(io, T_FILE, "IO", "to_io"); GetOpenFile(io, fptr); - fd = FPTR_TO_FD(fptr); + rb_io_check_closed(fptr); - if (fd < 0) - rb_raise(rb_eIOError, "closed stream"); - return fd; + return fptr->fd; } +#endif /* Ruby <3.1 !HAVE_RB_IO_DESCRIPTOR */ diff --git a/ext/raindrops/raindrops.c b/ext/raindrops/raindrops.c index 837084c..72a6ee7 100644 --- a/ext/raindrops/raindrops.c +++ b/ext/raindrops/raindrops.c @@ -4,6 +4,7 @@ #include <assert.h> #include <errno.h> #include <stddef.h> +#include <string.h> #include "raindrops_atomic.h" #ifndef SIZET2NUM @@ -34,10 +35,18 @@ struct raindrops { size_t size; size_t capa; pid_t pid; + VALUE io; struct raindrop *drops; }; /* called by GC */ +static void rd_mark(void *ptr) +{ + struct raindrops *r = ptr; + rb_gc_mark(r->io); +} + +/* called by GC */ static void rd_free(void *ptr) { struct raindrops *r = ptr; @@ -60,7 +69,7 @@ static size_t rd_memsize(const void *ptr) static const rb_data_type_t rd_type = { "raindrops", - { NULL, rd_free, rd_memsize, /* reserved */ }, + { rd_mark, rd_free, rd_memsize, /* reserved */ }, /* parent, data, [ flags ] */ }; @@ -87,16 +96,10 @@ static struct raindrops *get(VALUE self) } /* - * call-seq: - * Raindrops.new(size) -> raindrops object - * - * Initializes a Raindrops object to hold +size+ counters. +size+ is - * only a hint and the actual number of counters the object has is - * dependent on the CPU model, number of cores, and page size of - * the machine. The actual size of the object will always be equal - * or greater than the specified +size+. + * This is the _actual_ implementation of #initialize - the Ruby wrapper + * handles keyword-argument handling then calls this method. */ -static VALUE init(VALUE self, VALUE size) +static VALUE init_cimpl(VALUE self, VALUE size, VALUE io, VALUE zero) { struct raindrops *r = DATA_PTR(self); int tries = 1; @@ -113,9 +116,19 @@ static VALUE init(VALUE self, VALUE size) r->capa = tmp / raindrop_size; assert(PAGE_ALIGN(raindrop_size * r->capa) == tmp && "not aligned"); + r->io = io; + retry: - r->drops = mmap(NULL, tmp, - PROT_READ|PROT_WRITE, MAP_ANON|MAP_SHARED, -1, 0); + if (RTEST(r->io)) { + int fd = NUM2INT(rb_funcall(r->io, rb_intern("fileno"), 0)); + rb_funcall(r->io, rb_intern("truncate"), 1, SIZET2NUM(tmp)); + r->drops = mmap(NULL, tmp, + PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0); + } else { + r->drops = mmap(NULL, tmp, + PROT_READ|PROT_WRITE, MAP_ANON|MAP_SHARED, + -1, 0); + } if (r->drops == MAP_FAILED) { int err = errno; @@ -127,6 +140,9 @@ retry: } r->pid = getpid(); + if (RTEST(zero)) + memset(r->drops, 0, tmp); + return self; } @@ -217,14 +233,16 @@ static VALUE capa(VALUE self) * call-seq: * rd.dup -> rd_copy * - * Duplicates and snapshots the current state of a Raindrops object. + * Duplicates and snapshots the current state of a Raindrops object. Even + * if the given Raindrops object is backed by a file, the copy will be backed + * by independent, anonymously mapped memory. */ static VALUE init_copy(VALUE dest, VALUE source) { struct raindrops *dst = DATA_PTR(dest); struct raindrops *src = get(source); - init(dest, SIZET2NUM(src->size)); + init_cimpl(dest, SIZET2NUM(src->size), Qnil, Qfalse); memcpy(dst->drops, src->drops, raindrop_size * src->size); return dest; @@ -375,6 +393,20 @@ static VALUE evaporate_bang(VALUE self) return Qnil; } +/* + * call-seq: + * to_io -> IO + * + * Returns the IO object backing the memory for this raindrop, if + * one was specified when constructing this Raindrop. If this + * Raindrop is backed by anonymous memory, this method returns nil. + */ +static VALUE to_io(VALUE self) +{ + struct raindrops *r = get(self); + return r->io; +} + void Init_raindrops_ext(void) { VALUE cRaindrops = rb_define_class("Raindrops", rb_cObject); @@ -433,7 +465,7 @@ void Init_raindrops_ext(void) rb_define_alloc_func(cRaindrops, alloc); - rb_define_method(cRaindrops, "initialize", init, 1); + rb_define_private_method(cRaindrops, "initialize_cimpl", init_cimpl, 3); rb_define_method(cRaindrops, "incr", incr, -1); rb_define_method(cRaindrops, "decr", decr, -1); rb_define_method(cRaindrops, "to_ary", to_ary, 0); @@ -444,6 +476,7 @@ void Init_raindrops_ext(void) rb_define_method(cRaindrops, "capa", capa, 0); rb_define_method(cRaindrops, "initialize_copy", init_copy, 1); rb_define_method(cRaindrops, "evaporate!", evaporate_bang, 0); + rb_define_method(cRaindrops, "to_io", to_io, 0); #ifdef __linux__ Init_raindrops_linux_inet_diag(); diff --git a/ext/raindrops/tcp_info.c b/ext/raindrops/tcp_info.c index b82f705..c0d34e0 100644 --- a/ext/raindrops/tcp_info.c +++ b/ext/raindrops/tcp_info.c @@ -76,7 +76,7 @@ static VALUE alloc(VALUE klass) */ static VALUE init(VALUE self, VALUE io) { - int fd = my_fileno(io); + int fd = my_fileno(rb_io_get_io(io)); struct tcp_info *info = DATA_PTR(self); socklen_t len = (socklen_t)sizeof(struct tcp_info); int rc = getsockopt(fd, IPPROTO_TCP, TCP_INFO, info, &len); diff --git a/lib/raindrops.rb b/lib/raindrops.rb index ba273eb..c071d57 100644 --- a/lib/raindrops.rb +++ b/lib/raindrops.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false # # Each Raindrops object is a container that holds several counters. # It is internally a page-aligned, shared memory area that allows @@ -36,6 +37,30 @@ class Raindrops def total active + queued end + end unless defined? ListenStats + + # call-seq: + # Raindrops.new(size, io: nil) -> raindrops object + # + # Initializes a Raindrops object to hold +size+ counters. +size+ is + # only a hint and the actual number of counters the object has is + # dependent on the CPU model, number of cores, and page size of + # the machine. The actual size of the object will always be equal + # or greater than the specified +size+. + # If +io+ is provided, then the Raindrops memory will be backed by + # the specified file; otherwise, it will allocate anonymous memory. + # The IO object must respond to +truncate+, as this is used to set + # the size of the file. + # If +zero+ is provided, then the memory region is zeroed prior to + # returning. This is only meaningful if +io+ is also provided; in + # that case it controls whether any existing counter values in +io+ + # are retained (false) or whether it is entirely zeroed (true). + def initialize(size, io: nil, zero: false) + # This ruby wrapper exists to handle the keyword-argument handling, + # which is otherwise kind of awkward in C. We delegate the keyword + # arguments to the _actual_ initialize implementation as positional + # args. + initialize_cimpl(size, io, zero) end autoload :Linux, 'raindrops/linux' diff --git a/lib/raindrops/aggregate.rb b/lib/raindrops/aggregate.rb index 4fb731f..9ed7eb7 100644 --- a/lib/raindrops/aggregate.rb +++ b/lib/raindrops/aggregate.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false # # raindrops may use the {aggregate}[https://github.com/josephruscio/aggregate] # RubyGem to aggregate statistics from TCP_Info lookups. diff --git a/lib/raindrops/aggregate/last_data_recv.rb b/lib/raindrops/aggregate/last_data_recv.rb index 6919fbc..2205208 100644 --- a/lib/raindrops/aggregate/last_data_recv.rb +++ b/lib/raindrops/aggregate/last_data_recv.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "socket" # # @@ -10,6 +11,8 @@ require "socket" # Methods wrapped include: # - TCPServer#accept # - TCPServer#accept_nonblock +# - Socket#accept +# - Socket#accept_nonblock # - Kgio::TCPServer#kgio_accept # - Kgio::TCPServer#kgio_tryaccept module Raindrops::Aggregate::LastDataRecv @@ -33,8 +36,10 @@ module Raindrops::Aggregate::LastDataRecv # automatically extends any TCPServer objects used by Unicorn def self.cornify! - Unicorn::HttpServer::LISTENERS.each do |sock| - sock.extend(self) if TCPServer === sock + Unicorn::HttpServer::LISTENERS.each do |s| + if TCPServer === s || (s.instance_of?(Socket) && s.local_address.ip?) + s.extend(self) + end end end @@ -60,8 +65,8 @@ module Raindrops::Aggregate::LastDataRecv count! super end - def accept_nonblock - count! super + def accept_nonblock(exception: true) + count! super(exception: exception) end # :startdoc: @@ -72,12 +77,19 @@ module Raindrops::Aggregate::LastDataRecv # # We require TCP_DEFER_ACCEPT on the listen socket for # +last_data_recv+ to be accurate - def count!(io) + def count!(ret) + case ret + when :wait_readable + when Array # Socket#accept_nonblock + io = ret[0] + else # TCPSocket#accept_nonblock + io = ret + end if io x = Raindrops::TCP_Info.new(io) @raindrops_aggregate << x.last_data_recv end - io + ret end end diff --git a/lib/raindrops/aggregate/pmq.rb b/lib/raindrops/aggregate/pmq.rb index 64d0a4f..94bdf4f 100644 --- a/lib/raindrops/aggregate/pmq.rb +++ b/lib/raindrops/aggregate/pmq.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "tempfile" require "aggregate" require "posix_mq" diff --git a/lib/raindrops/last_data_recv.rb b/lib/raindrops/last_data_recv.rb index b4808a1..e6c47e1 100644 --- a/lib/raindrops/last_data_recv.rb +++ b/lib/raindrops/last_data_recv.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "raindrops" # This is highly experimental! diff --git a/lib/raindrops/linux.rb b/lib/raindrops/linux.rb index 9842ae1..a76192c 100644 --- a/lib/raindrops/linux.rb +++ b/lib/raindrops/linux.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false # For reporting TCP ListenStats, users of older \Linux kernels need to ensure # that the the "inet_diag" and "tcp_diag" kernel modules are loaded as they do diff --git a/lib/raindrops/middleware.rb b/lib/raindrops/middleware.rb index d5e3927..25b5a1e 100644 --- a/lib/raindrops/middleware.rb +++ b/lib/raindrops/middleware.rb @@ -1,5 +1,7 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'raindrops' +require 'thread' # Raindrops::Middleware is Rack middleware that allows snapshotting # current activity from an HTTP request. For all operating systems, @@ -93,11 +95,12 @@ class Raindrops::Middleware @app = app @stats = opts[:stats] || Stats.new @path = opts[:path] || "/_raindrops" + @mtx = Mutex.new tmp = opts[:listeners] if tmp.nil? && defined?(Unicorn) && Unicorn.respond_to?(:listener_names) tmp = Unicorn.listener_names end - @tcp = @unix = nil + @nl_sock = @tcp = @unix = nil if tmp @tcp = tmp.grep(/\A.+:\d+\z/) @@ -129,9 +132,12 @@ class Raindrops::Middleware "writing: #{@stats.writing}\n" if defined?(Raindrops::Linux.tcp_listener_stats) - Raindrops::Linux.tcp_listener_stats(@tcp).each do |addr,stats| - body << "#{addr} active: #{stats.active}\n" \ - "#{addr} queued: #{stats.queued}\n" + @mtx.synchronize do + @nl_sock ||= Raindrops::InetDiagSocket.new + Raindrops::Linux.tcp_listener_stats(@tcp, @nl_sock).each do |addr,stats| + body << "#{addr} active: #{stats.active}\n" \ + "#{addr} queued: #{stats.queued}\n" + end end if @tcp Raindrops::Linux.unix_listener_stats(@unix).each do |addr,stats| body << "#{addr} active: #{stats.active}\n" \ diff --git a/lib/raindrops/middleware/proxy.rb b/lib/raindrops/middleware/proxy.rb index a7c8e66..433950c 100644 --- a/lib/raindrops/middleware/proxy.rb +++ b/lib/raindrops/middleware/proxy.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false # :stopdoc: # This class is used by Raindrops::Middleware to proxy application # response bodies. There should be no need to use it directly. diff --git a/lib/raindrops/struct.rb b/lib/raindrops/struct.rb index e81a78e..7233ce8 100644 --- a/lib/raindrops/struct.rb +++ b/lib/raindrops/struct.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false # This is a wrapper around Raindrops objects much like the core Ruby # \Struct can be seen as a wrapper around the core \Array class. diff --git a/lib/raindrops/watcher.rb b/lib/raindrops/watcher.rb index ac5b895..8fc0772 100644 --- a/lib/raindrops/watcher.rb +++ b/lib/raindrops/watcher.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "thread" require "time" require "socket" diff --git a/raindrops.gemspec b/raindrops.gemspec index 1de56a0..2f171db 100644 --- a/raindrops.gemspec +++ b/raindrops.gemspec @@ -21,6 +21,6 @@ Gem::Specification.new do |s| s.add_development_dependency('aggregate', '~> 0.2') s.add_development_dependency('test-unit', '~> 3.0') s.add_development_dependency('posix_mq', '~> 2.0') - s.add_development_dependency('rack', [ '>= 1.2', '< 3.0' ]) + s.add_development_dependency('rack', [ '>= 1.2', '< 4' ]) s.licenses = %w(LGPL-2.1+) end @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false # # setup.rb # diff --git a/test/ipv6_enabled.rb b/test/ipv6_enabled.rb index c4c9709..84ed9c1 100644 --- a/test/ipv6_enabled.rb +++ b/test/ipv6_enabled.rb @@ -1,3 +1,4 @@ +# frozen_string_literal: false def ipv6_enabled? tmp = TCPServer.new(ENV["TEST_HOST6"] || '::1', 0) tmp.close diff --git a/test/rack_unicorn.rb b/test/rack_unicorn.rb index 779e8bf..05a7751 100644 --- a/test/rack_unicorn.rb +++ b/test/rack_unicorn.rb @@ -1,11 +1,11 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "test/unit" require "raindrops" -require "rack" -require "rack/lobster" require "open-uri" begin require "unicorn" + require "rack" require "rack/lobster" rescue LoadError => e warn "W: #{e} skipping test since Rack or Unicorn was not found" diff --git a/test/test_aggregate_pmq.rb b/test/test_aggregate_pmq.rb index 692b9bd..24e0277 100644 --- a/test/test_aggregate_pmq.rb +++ b/test/test_aggregate_pmq.rb @@ -1,3 +1,4 @@ +# frozen_string_literal: false require "test/unit" require "raindrops" pmq = begin diff --git a/test/test_inet_diag_socket.rb b/test/test_inet_diag_socket.rb index a8c9973..e310dff 100644 --- a/test/test_inet_diag_socket.rb +++ b/test/test_inet_diag_socket.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'raindrops' require 'fcntl' diff --git a/test/test_last_data_recv.rb b/test/test_last_data_recv.rb new file mode 100644 index 0000000..edd00f3 --- /dev/null +++ b/test/test_last_data_recv.rb @@ -0,0 +1,58 @@ +# frozen_string_literal: false +begin + require 'aggregate' + have_aggregate = true +rescue LoadError => e + warn "W: #{e} skipping #{__FILE__}" +end +require 'test/unit' +require 'raindrops' +require 'io/wait' + +class TestLastDataRecv < Test::Unit::TestCase + def setup + Raindrops::Aggregate::LastDataRecv.default_aggregate = [] + end + + def teardown + Raindrops::Aggregate::LastDataRecv.default_aggregate = nil + end + + def test_accept_nonblock_agg + s = Socket.new(:INET, :STREAM, 0) + s.listen(128) + addr = s.connect_address + s.extend(Raindrops::Aggregate::LastDataRecv) + s.raindrops_aggregate = [] + c = Socket.new(:INET, :STREAM, 0) + c.connect(addr) + c.write '.' # for TCP_DEFER_ACCEPT + client, ai = s.accept_nonblock(exception: false) + assert client.kind_of?(Socket) + assert ai.kind_of?(Addrinfo) + assert_equal 1, s.raindrops_aggregate.size + assert s.raindrops_aggregate[0].instance_of?(Integer) + client, ai = s.accept_nonblock(exception: false) + assert_equal :wait_readable, client + assert_nil ai + assert_equal 1, s.raindrops_aggregate.size + assert_raise(IO::WaitReadable) { s.accept_nonblock } + end + + def test_accept_nonblock_one + s = TCPServer.new('127.0.0.1', 0) + s.extend(Raindrops::Aggregate::LastDataRecv) + s.raindrops_aggregate = [] + addr = s.addr + c = TCPSocket.new(addr[3], addr[1]) + c.write '.' # for TCP_DEFER_ACCEPT + client = s.accept_nonblock(exception: false) + assert client.kind_of?(TCPSocket) + assert_equal 1, s.raindrops_aggregate.size + assert s.raindrops_aggregate[0].instance_of?(Integer) + client = s.accept_nonblock(exception: false) + assert_equal :wait_readable, client + assert_equal 1, s.raindrops_aggregate.size + assert_raise(IO::WaitReadable) { s.accept_nonblock } + end +end if RUBY_PLATFORM =~ /linux/ && have_aggregate diff --git a/test/test_last_data_recv_unicorn.rb b/test/test_last_data_recv_unicorn.rb index 60d1be9..55f5e7f 100644 --- a/test/test_last_data_recv_unicorn.rb +++ b/test/test_last_data_recv_unicorn.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "./test/rack_unicorn" require "tempfile" require "net/http" diff --git a/test/test_linux.rb b/test/test_linux.rb index 7808469..5451c3f 100644 --- a/test/test_linux.rb +++ b/test/test_linux.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'tempfile' require 'raindrops' diff --git a/test/test_linux_all_tcp_listen_stats.rb b/test/test_linux_all_tcp_listen_stats.rb index ef1f943..12a35ba 100644 --- a/test/test_linux_all_tcp_listen_stats.rb +++ b/test/test_linux_all_tcp_listen_stats.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'socket' require 'raindrops' diff --git a/test/test_linux_all_tcp_listen_stats_leak.rb b/test/test_linux_all_tcp_listen_stats_leak.rb index 7be46d4..a3da07e 100644 --- a/test/test_linux_all_tcp_listen_stats_leak.rb +++ b/test/test_linux_all_tcp_listen_stats_leak.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'raindrops' require 'socket' diff --git a/test/test_linux_ipv6.rb b/test/test_linux_ipv6.rb index 9e8730a..9ef8f0a 100644 --- a/test/test_linux_ipv6.rb +++ b/test/test_linux_ipv6.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'tempfile' require 'raindrops' diff --git a/test/test_linux_middleware.rb b/test/test_linux_middleware.rb index f573225..7ed20df 100644 --- a/test/test_linux_middleware.rb +++ b/test/test_linux_middleware.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'tempfile' require 'raindrops' diff --git a/test/test_linux_reuseport_tcp_listen_stats.rb b/test/test_linux_reuseport_tcp_listen_stats.rb new file mode 100644 index 0000000..82083e0 --- /dev/null +++ b/test/test_linux_reuseport_tcp_listen_stats.rb @@ -0,0 +1,52 @@ +# -*- encoding: binary -*- +# frozen_string_literal: false +require "./test/rack_unicorn" +require 'test/unit' +require 'socket' +require 'raindrops' +$stderr.sync = $stdout.sync = true + +class TestLinuxReuseportTcpListenStats < Test::Unit::TestCase + include Raindrops::Linux + include Unicorn::SocketHelper + TEST_ADDR = ENV['UNICORN_TEST_ADDR'] || '127.0.0.1' + DEFAULT_BACKLOG = 10 + + def setup + @socks = [] + end + + def teardown + @socks.each { |io| io.closed? or io.close } + end + + def new_socket_server(**kwargs) + s = new_tcp_server TEST_ADDR, kwargs[:port] || 0, kwargs + s.listen(kwargs[:backlog] || DEFAULT_BACKLOG) + @socks << s + [ s, s.addr[1] ] + end + + def new_client(port) + s = TCPSocket.new("127.0.0.1", port) + @socks << s + s + end + + def test_reuseport_queue_stats + listeners = 10 + _, port = new_socket_server(reuseport: true) + addr = "#{TEST_ADDR}:#{port}" + (listeners - 1).times do + new_socket_server(reuseport: true, port: port) + end + + listeners.times do |i| + all = Raindrops::Linux.tcp_listener_stats + assert_equal [0, i], all[addr].to_a + new_client(port) + all = Raindrops::Linux.tcp_listener_stats + assert_equal [0, i+1], all[addr].to_a + end + end +end if RUBY_PLATFORM =~ /linux/ && Object.const_defined?(:Unicorn) diff --git a/test/test_middleware.rb b/test/test_middleware.rb index 56ce346..5694cd4 100644 --- a/test/test_middleware.rb +++ b/test/test_middleware.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'raindrops' diff --git a/test/test_middleware_unicorn.rb b/test/test_middleware_unicorn.rb index 6730d4b..53226a9 100644 --- a/test/test_middleware_unicorn.rb +++ b/test/test_middleware_unicorn.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "./test/rack_unicorn" $stderr.sync = $stdout.sync = true diff --git a/test/test_middleware_unicorn_ipv6.rb b/test/test_middleware_unicorn_ipv6.rb index 3d6862c..99ecb7f 100644 --- a/test/test_middleware_unicorn_ipv6.rb +++ b/test/test_middleware_unicorn_ipv6.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "./test/rack_unicorn" require "./test/ipv6_enabled" $stderr.sync = $stdout.sync = true diff --git a/test/test_raindrops.rb b/test/test_raindrops.rb index 0749694..165766e 100644 --- a/test/test_raindrops.rb +++ b/test/test_raindrops.rb @@ -1,6 +1,8 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'raindrops' +require 'tempfile' class TestRaindrops < Test::Unit::TestCase @@ -162,4 +164,45 @@ class TestRaindrops < Test::Unit::TestCase assert status.success? assert_equal [ 1, 2 ], tmp.to_ary end + + def test_io_backed + file = Tempfile.new('test_io_backed') + rd = Raindrops.new(4, io: file, zero: true) + rd[0] = 123 + rd[1] = 456 + + assert_equal 123, rd[0] + assert_equal 456, rd[1] + + rd.evaporate! + + file.rewind + data = file.read + assert_equal 123, data.unpack('L!')[0] + assert_equal 456, data[Raindrops::SIZE..data.size].unpack('L!')[0] + end + + def test_io_backed_reuse + file = Tempfile.new('test_io_backed') + rd = Raindrops.new(4, io: file, zero: true) + rd[0] = 123 + rd[1] = 456 + rd.evaporate! + + rd = Raindrops.new(4, io: file, zero: false) + assert_equal 123, rd[0] + assert_equal 456, rd[1] + end + + def test_iobacked_noreuse + file = Tempfile.new('test_io_backed') + rd = Raindrops.new(4, io: file, zero: true) + rd[0] = 123 + rd[1] = 456 + rd.evaporate! + + rd = Raindrops.new(4, io: file, zero: true) + assert_equal 0, rd[0] + assert_equal 0, rd[1] + end end diff --git a/test/test_raindrops_gc.rb b/test/test_raindrops_gc.rb index 2098129..a9f2026 100644 --- a/test/test_raindrops_gc.rb +++ b/test/test_raindrops_gc.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'raindrops' diff --git a/test/test_struct.rb b/test/test_struct.rb index 9792d5b..abf0c59 100644 --- a/test/test_struct.rb +++ b/test/test_struct.rb @@ -1,3 +1,4 @@ +# frozen_string_literal: false require 'test/unit' require 'raindrops' diff --git a/test/test_tcp_info.rb b/test/test_tcp_info.rb index 2ddacfd..2dc5c50 100644 --- a/test/test_tcp_info.rb +++ b/test/test_tcp_info.rb @@ -1,4 +1,5 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require 'test/unit' require 'tempfile' require 'raindrops' diff --git a/test/test_watcher.rb b/test/test_watcher.rb index 28ac49b..3cf667c 100644 --- a/test/test_watcher.rb +++ b/test/test_watcher.rb @@ -1,9 +1,10 @@ # -*- encoding: binary -*- +# frozen_string_literal: false require "test/unit" -require "rack" require "raindrops" begin require 'aggregate' + require 'rack' rescue LoadError => e warn "W: #{e} skipping #{__FILE__}" end @@ -183,4 +184,4 @@ class TestWatcher < Test::Unit::TestCase assert_equal queued_before, headers["X-Last-Peak-At"], "should not change" assert_equal start, headers["X-First-Peak-At"] end -end if RUBY_PLATFORM =~ /linux/ && defined?(Aggregate) +end if RUBY_PLATFORM =~ /linux/ && defined?(Aggregate) && defined?(Rack) |