[sheepdog] [PATCH v8 1/3] lib: move sockfd cache from sheep to lib
Hitoshi Mitake
mitake.hitoshi at lab.ntt.co.jp
Wed Jul 31 09:11:46 CEST 2013
On some subcommands, collie also issues lots of request to sheeps. So
collie can enjoy sockfd caching. For this purpose, this patch moves
sockfd from sheep to lib and generalize the interfaces of sockfd.
This patch doesn't change anything related to IO NIC in sockfd
cache. Because collie can know address and port of IO NIC so
potentially use them.
Signed-off-by: Hitoshi Mitake <mitake.hitoshi at lab.ntt.co.jp>
---
v8:
- rename sockfd_node_del() -> sockfd_cache_del_node() for consistent
naming
v7:
- fix a mistake in sockfd_cache_del() made by the previous patch
v6:
- revive revalidate_node() because it is required for handling corner
case of sheep and doesn't harm collie's behavior
- trivial cleaning
v5:
- make the workqueue for cachefd growth an internal thing in
lib/sockfd_cache.c
- remove revalidate_node(), because this relies on the assumption of
sheep
- better naming of functions
- misc cleanings
include/Makefile.am | 3 +-
include/internal_proto.h | 2 +
include/sockfd_cache.h | 22 ++
lib/Makefile.am | 2 +-
{sheep => lib}/sockfd_cache.c | 125 +++++-----
sheep/gateway.c | 10 +-
sheep/group.c | 2 +-
sheep/sheep.c | 9 +-
sheep/sheep_priv.h | 12 +-
sheep/sockfd_cache.c | 509 +----------------------------------------
10 files changed, 105 insertions(+), 591 deletions(-)
create mode 100644 include/sockfd_cache.h
copy {sheep => lib}/sockfd_cache.c (89%)
diff --git a/include/Makefile.am b/include/Makefile.am
index 0acb76e..4d0c229 100644
--- a/include/Makefile.am
+++ b/include/Makefile.am
@@ -2,4 +2,5 @@ MAINTAINERCLEANFILES = Makefile.in config.h.in
noinst_HEADERS = bitops.h event.h logger.h sheepdog_proto.h util.h \
list.h net.h sheep.h exits.h strbuf.h rbtree.h \
- sha1.h option.h internal_proto.h shepherd.h work.h
+ sha1.h option.h internal_proto.h shepherd.h work.h \
+ sockfd_cache.h
diff --git a/include/internal_proto.h b/include/internal_proto.h
index 0463eae..0061007 100644
--- a/include/internal_proto.h
+++ b/include/internal_proto.h
@@ -20,6 +20,8 @@
#include <stdint.h>
#include <netinet/in.h>
+#include "sheepdog_proto.h"
+
#define SD_SHEEP_PROTO_VER 0x08
#define SD_DEFAULT_COPIES 3
diff --git a/include/sockfd_cache.h b/include/sockfd_cache.h
new file mode 100644
index 0000000..d91c56a
--- /dev/null
+++ b/include/sockfd_cache.h
@@ -0,0 +1,22 @@
+#ifndef SOCKFD_CACHE_H
+#define SOCKFD_CACHE_H
+
+#include "internal_proto.h"
+#include "work.h"
+
+struct sockfd *sockfd_cache_get(const struct node_id *nid);
+void sockfd_cache_put(const struct node_id *nid, struct sockfd *sfd);
+void sockfd_cache_del_node(const struct node_id *nid);
+void sockfd_cache_del(const struct node_id *nid, struct sockfd *sfd);
+void sockfd_cache_add(const struct node_id *nid);
+void sockfd_cache_add_group(const struct sd_node *nodes, int nr);
+
+int sockfd_init(void);
+
+/* sockfd_cache */
+struct sockfd {
+ int fd;
+ int idx;
+};
+
+#endif /* SOCKFD_CACHE_H */
diff --git a/lib/Makefile.am b/lib/Makefile.am
index a95ce4c..22edf6e 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -5,7 +5,7 @@ INCLUDES = -I$(top_builddir)/include -I$(top_srcdir)/include
noinst_LIBRARIES = libsheepdog.a
libsheepdog_a_SOURCES = event.c logger.c net.c util.c rbtree.c strbuf.c \
- sha1.c option.c work.c
+ sha1.c option.c work.c sockfd_cache.c
# support for GNU Flymake
check-syntax:
diff --git a/sheep/sockfd_cache.c b/lib/sockfd_cache.c
similarity index 89%
copy from sheep/sockfd_cache.c
copy to lib/sockfd_cache.c
index 13bb7f6..98cea94 100644
--- a/sheep/sockfd_cache.c
+++ b/lib/sockfd_cache.c
@@ -27,7 +27,13 @@
* 7 support dual connections to a single node.
*/
-#include "sheep_priv.h"
+#include <pthread.h>
+
+#include "sockfd_cache.h"
+#include "work.h"
+#include "rbtree.h"
+#include "util.h"
+#include "sheep.h"
struct sockfd_cache {
struct rb_root root;
@@ -208,20 +214,6 @@ false_out:
return false;
}
-/* When node craches, we should delete it from the cache */
-void sockfd_cache_del(const struct node_id *nid)
-{
- char name[INET6_ADDRSTRLEN];
- int n;
-
- if (!sockfd_cache_destroy(nid))
- return;
-
- n = uatomic_sub_return(&sockfd_cache.count, 1);
- addr_to_str(name, sizeof(name), nid->addr, 0);
- sd_dprintf("%s:%d, count %d", name, nid->port, n);
-}
-
static void sockfd_cache_add_nolock(const struct node_id *nid)
{
struct sockfd_cache_entry *new = xmalloc(sizeof(*new));
@@ -282,6 +274,8 @@ void sockfd_cache_add(const struct node_id *nid)
static uatomic_bool fds_in_grow;
static int fds_high_watermark = FDS_WATERMARK(DEFAULT_FDS_COUNT);
+static struct work_queue *grow_wq;
+
static void do_grow_fds(struct work *work)
{
struct sockfd_cache_entry *entry;
@@ -318,6 +312,15 @@ static inline void check_idx(int idx)
{
struct work *w;
+ if (!grow_wq) {
+ sd_dprintf("Request for growing fds is issued, but"
+ " a work queue for this work is not registered yet.");
+ sd_dprintf("Initializing work queue in earlier stage"
+ " is suggested.");
+
+ return;
+ }
+
if (idx <= fds_high_watermark)
return;
if (!uatomic_set_true(&fds_in_grow))
@@ -326,7 +329,7 @@ static inline void check_idx(int idx)
w = xmalloc(sizeof(*w));
w->fn = do_grow_fds;
w->done = grow_fds_done;
- queue_work(sys->sockfd_wqueue, w);
+ queue_work(grow_wq, w);
}
/* Add the node back if it is still alive */
@@ -350,7 +353,7 @@ alive:
}
/* Try to create/get cached IO connection. If failed, fallback to non-IO one */
-static struct sockfd *sockfd_cache_get(const struct node_id *nid)
+static struct sockfd *sockfd_cache_get_long(const struct node_id *nid)
{
struct sockfd_cache_entry *entry;
struct sockfd *sfd;
@@ -372,8 +375,10 @@ grab:
*/
if (!revalidate_node(nid))
return NULL;
+
goto grab;
}
+
check_idx(idx);
if (entry->fds[idx].fd != -1) {
sd_dprintf("%s:%d, idx %d", name, port, idx);
@@ -402,7 +407,7 @@ out:
return sfd;
}
-static void sockfd_cache_put(const struct node_id *nid, int idx)
+static void sockfd_cache_put_long(const struct node_id *nid, int idx)
{
bool use_io = nid->io_port ? true : false;
const uint8_t *addr = use_io ? nid->io_addr : nid->addr;
@@ -442,20 +447,36 @@ static void sockfd_cache_close(const struct node_id *nid, int idx)
}
/*
+ * Create work queue for growing fds.
+ * Before this function called, growing cannot be done.
+ */
+int sockfd_init(void)
+{
+ grow_wq = create_ordered_work_queue("sockfd_grow");
+
+ if (!grow_wq) {
+ sd_eprintf("error at creating workqueue for sockfd growth");
+ return -1;
+ }
+
+ return 0;
+}
+
+/*
* Return a sockfd connected to the node to the caller
*
* Try to get a 'long' FD as best, which is cached and never closed. If no FD
* available, we return a 'short' FD which is supposed to be closed by
- * sheep_put_sockfd().
+ * put_sockfd().
*
* ret_idx is opaque to the caller, -1 indicates it is a short FD.
*/
-struct sockfd *sheep_get_sockfd(const struct node_id *nid)
+struct sockfd *sockfd_cache_get(const struct node_id *nid)
{
struct sockfd *sfd;
int fd;
- sfd = sockfd_cache_get(nid);
+ sfd = sockfd_cache_get_long(nid);
if (sfd)
return sfd;
@@ -473,16 +494,12 @@ struct sockfd *sheep_get_sockfd(const struct node_id *nid)
/*
* Release a sockfd connected to the node, which is acquired from
- * sheep_get_sockfd()
+ * sockfd_cache_get()
*
* If it is a long FD, just decrease the refcount to make it available again.
* If it is a short FD, close it.
- *
- * sheep_put_sockfd() or sheep_del_sockfd() should be paired with
- * sheep_get_sockfd()
*/
-
-void sheep_put_sockfd(const struct node_id *nid, struct sockfd *sfd)
+void sockfd_cache_put(const struct node_id *nid, struct sockfd *sfd)
{
if (sfd->idx == -1) {
sd_dprintf("%d", sfd->fd);
@@ -491,18 +508,32 @@ void sheep_put_sockfd(const struct node_id *nid, struct sockfd *sfd)
return;
}
- sockfd_cache_put(nid, sfd->idx);
+ sockfd_cache_put_long(nid, sfd->idx);
free(sfd);
}
+/* Delete all sockfd connected to the node, when node is crashed. */
+void sockfd_cache_del_node(const struct node_id *nid)
+{
+ char name[INET6_ADDRSTRLEN];
+ int n;
+
+ if (!sockfd_cache_destroy(nid))
+ return;
+
+ n = uatomic_sub_return(&sockfd_cache.count, 1);
+ addr_to_str(name, sizeof(name), nid->addr, 0);
+ sd_dprintf("%s:%d, count %d", name, nid->port, n);
+}
+
/*
- * Delete a sockfd connected to the node, when node is crashed.
+ * Delete a sockfd connected to the node.
*
* If it is a long FD, de-refcount it and tres to destroy all the cached FDs of
* this node in the cache.
* If it is a short FD, just close it.
*/
-void sheep_del_sockfd(const struct node_id *nid, struct sockfd *sfd)
+void sockfd_cache_del(const struct node_id *nid, struct sockfd *sfd)
{
if (sfd->idx == -1) {
sd_dprintf("%d", sfd->fd);
@@ -512,38 +543,6 @@ void sheep_del_sockfd(const struct node_id *nid, struct sockfd *sfd)
}
sockfd_cache_close(nid, sfd->idx);
- sockfd_cache_del(nid);
+ sockfd_cache_del_node(nid);
free(sfd);
}
-
-int sheep_exec_req(const struct node_id *nid, struct sd_req *hdr, void *buf)
-{
- struct sd_rsp *rsp = (struct sd_rsp *)hdr;
- struct sockfd *sfd;
- int ret;
-
- assert(is_worker_thread());
-
- sfd = sheep_get_sockfd(nid);
- if (!sfd)
- return SD_RES_NETWORK_ERROR;
-
- ret = exec_req(sfd->fd, hdr, buf, sheep_need_retry, hdr->epoch,
- MAX_RETRY_COUNT);
- if (ret) {
- sd_dprintf("remote node might have gone away");
- sheep_del_sockfd(nid, sfd);
- return SD_RES_NETWORK_ERROR;
- }
- ret = rsp->result;
- if (ret != SD_RES_SUCCESS)
- sd_eprintf("failed %s", sd_strerror(ret));
-
- sheep_put_sockfd(nid, sfd);
- return ret;
-}
-
-bool sheep_need_retry(uint32_t epoch)
-{
- return sys_epoch() == epoch;
-}
diff --git a/sheep/gateway.c b/sheep/gateway.c
index 59d0127..fb7e9a7 100644
--- a/sheep/gateway.c
+++ b/sheep/gateway.c
@@ -114,13 +114,13 @@ static inline void write_info_update(struct write_info *wi, int pos)
static inline void finish_one_write(struct write_info *wi, int i)
{
- sheep_put_sockfd(wi->ent[i].nid, wi->ent[i].sfd);
+ sockfd_cache_put(wi->ent[i].nid, wi->ent[i].sfd);
write_info_update(wi, i);
}
static inline void finish_one_write_err(struct write_info *wi, int i)
{
- sheep_del_sockfd(wi->ent[i].nid, wi->ent[i].sfd);
+ sockfd_cache_del(wi->ent[i].nid, wi->ent[i].sfd);
write_info_update(wi, i);
}
@@ -176,7 +176,7 @@ again:
nr_sent = wi->nr_sent;
/* XXX Blinedly close all the connections */
for (i = 0; i < nr_sent; i++)
- sheep_del_sockfd(wi->ent[i].nid, wi->ent[i].sfd);
+ sockfd_cache_del(wi->ent[i].nid, wi->ent[i].sfd);
return SD_RES_NETWORK_ERROR;
}
@@ -278,7 +278,7 @@ static int gateway_forward_request(struct request *req)
}
nid = &target_nodes[i]->nid;
- sfd = sheep_get_sockfd(nid);
+ sfd = sockfd_cache_get(nid);
if (!sfd) {
err_ret = SD_RES_NETWORK_ERROR;
break;
@@ -288,7 +288,7 @@ static int gateway_forward_request(struct request *req)
sheep_need_retry, req->rq.epoch,
MAX_RETRY_COUNT);
if (ret) {
- sheep_del_sockfd(nid, sfd);
+ sockfd_cache_del_node(nid);
err_ret = SD_RES_NETWORK_ERROR;
sd_dprintf("fail %d", ret);
break;
diff --git a/sheep/group.c b/sheep/group.c
index 011a4fa..474e48a 100644
--- a/sheep/group.c
+++ b/sheep/group.c
@@ -998,7 +998,7 @@ void sd_leave_handler(const struct sd_node *left, const struct sd_node *members,
put_vnode_info(old_vnode_info);
- sockfd_cache_del(&left->nid);
+ sockfd_cache_del_node(&left->nid);
}
static void update_node_size(struct sd_node *node)
diff --git a/sheep/sheep.c b/sheep/sheep.c
index c8d4478..39288a8 100644
--- a/sheep/sheep.c
+++ b/sheep/sheep.c
@@ -430,7 +430,6 @@ static int create_work_queues(void)
sys->recovery_wqueue = create_ordered_work_queue("rw");
sys->deletion_wqueue = create_ordered_work_queue("deletion");
sys->block_wqueue = create_ordered_work_queue("block");
- sys->sockfd_wqueue = create_ordered_work_queue("sockfd");
sys->md_wqueue = create_ordered_work_queue("md");
if (sys->enable_object_cache) {
sys->oc_reclaim_wqueue =
@@ -440,9 +439,9 @@ static int create_work_queues(void)
return -1;
}
if (!sys->gateway_wqueue || !sys->io_wqueue || !sys->recovery_wqueue ||
- !sys->deletion_wqueue || !sys->block_wqueue ||
- !sys->sockfd_wqueue || !sys->md_wqueue)
+ !sys->deletion_wqueue || !sys->block_wqueue || !sys->md_wqueue)
return -1;
+
return 0;
}
@@ -835,6 +834,10 @@ int main(int argc, char **argv)
if (ret)
exit(1);
+ ret = sockfd_init();
+ if (ret)
+ exit(1);
+
ret = init_store_driver(sys->gateway_only);
if (ret)
exit(1);
diff --git a/sheep/sheep_priv.h b/sheep/sheep_priv.h
index bc95cd1..09303ff 100644
--- a/sheep/sheep_priv.h
+++ b/sheep/sheep_priv.h
@@ -42,6 +42,7 @@
#include "strbuf.h"
#include "sha1.h"
#include "config.h"
+#include "sockfd_cache.h"
struct client_info {
struct connection conn;
@@ -116,7 +117,6 @@ struct system_info {
struct work_queue *recovery_wqueue;
struct work_queue *recovery_notify_wqueue;
struct work_queue *block_wqueue;
- struct work_queue *sockfd_wqueue;
struct work_queue *oc_reclaim_wqueue;
struct work_queue *oc_push_wqueue;
struct work_queue *md_wqueue;
@@ -399,16 +399,6 @@ int object_cache_remove(uint64_t oid);
/* store layout migration */
int sd_migrate_store(int from, int to);
-/* sockfd_cache */
-struct sockfd {
- int fd;
- int idx;
-};
-
-void sockfd_cache_del(const struct node_id *);
-void sockfd_cache_add(const struct node_id *);
-void sockfd_cache_add_group(const struct sd_node *nodes, int nr);
-
struct sockfd *sheep_get_sockfd(const struct node_id *);
void sheep_put_sockfd(const struct node_id *, struct sockfd *);
void sheep_del_sockfd(const struct node_id *, struct sockfd *);
diff --git a/sheep/sockfd_cache.c b/sheep/sockfd_cache.c
index 13bb7f6..20bca58 100644
--- a/sheep/sockfd_cache.c
+++ b/sheep/sockfd_cache.c
@@ -11,511 +11,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
-/*
- * The sockfd cache provides us long TCP connections connected to the nodes
- * in the cluster to accerlater the data transfer, which has the following
- * characteristics:
- * 0 dynamically allocated/deallocated at node granularity.
- * 1 cached fds are multiplexed by all threads.
- * 2 each session (for e.g, forward_write_obj_req) can grab one fd at a time.
- * 3 if there isn't any FD available from cache, use normal connect_to() and
- * close() internally.
- * 4 FD are named by IP:PORT uniquely, hence no need of resetting at
- * membership change.
- * 5 the total number of FDs is scalable to massive nodes.
- * 6 total 3 APIs: sheep_{get,put,del}_sockfd().
- * 7 support dual connections to a single node.
- */
-
#include "sheep_priv.h"
-struct sockfd_cache {
- struct rb_root root;
- pthread_rwlock_t lock;
- int count;
-};
-
-static struct sockfd_cache sockfd_cache = {
- .root = RB_ROOT,
- .lock = PTHREAD_RWLOCK_INITIALIZER,
-};
-
-/*
- * Suppose request size from Guest is 512k, then 4M / 512k = 8, so at
- * most 8 requests can be issued to the same sheep object. Based on this
- * assumption, '8' would be effecient for servers that only host 2~4
- * Guests.
- *
- * This fd count will be dynamically grown when the idx reaches watermark which
- * is calculated by FDS_WATERMARK
- */
-#define FDS_WATERMARK(x) ((x) * 3 / 4)
-#define DEFAULT_FDS_COUNT 8
-
-/* How many FDs we cache for one node */
-static int fds_count = DEFAULT_FDS_COUNT;
-
-struct sockfd_cache_fd {
- int fd;
- uatomic_bool in_use;
-};
-
-struct sockfd_cache_entry {
- struct rb_node rb;
- struct node_id nid;
- struct sockfd_cache_fd *fds;
-};
-
-static struct sockfd_cache_entry *
-sockfd_cache_insert(struct sockfd_cache_entry *new)
-{
- struct rb_node **p = &sockfd_cache.root.rb_node;
- struct rb_node *parent = NULL;
- struct sockfd_cache_entry *entry;
-
- while (*p) {
- int cmp;
-
- parent = *p;
- entry = rb_entry(parent, struct sockfd_cache_entry, rb);
- cmp = node_id_cmp(&new->nid, &entry->nid);
-
- if (cmp < 0)
- p = &(*p)->rb_left;
- else if (cmp > 0)
- p = &(*p)->rb_right;
- else
- return entry;
- }
- rb_link_node(&new->rb, parent, p);
- rb_insert_color(&new->rb, &sockfd_cache.root);
-
- return NULL; /* insert successfully */
-}
-
-static struct sockfd_cache_entry *sockfd_cache_search(const struct node_id *nid)
-{
- struct rb_node *n = sockfd_cache.root.rb_node;
- struct sockfd_cache_entry *t;
-
- while (n) {
- int cmp;
-
- t = rb_entry(n, struct sockfd_cache_entry, rb);
- cmp = node_id_cmp(nid, &t->nid);
-
- if (cmp < 0)
- n = n->rb_left;
- else if (cmp > 0)
- n = n->rb_right;
- else
- return t; /* found it */
- }
-
- return NULL;
-}
-
-static inline int get_free_slot(struct sockfd_cache_entry *entry)
-{
- int idx = -1, i;
-
- for (i = 0; i < fds_count; i++) {
- if (!uatomic_set_true(&entry->fds[i].in_use))
- continue;
- idx = i;
- break;
- }
- return idx;
-}
-
-/*
- * Grab a free slot of the node and inc the refcount of the slot
- *
- * If no free slot available, this typically means we should use short FD.
- */
-static struct sockfd_cache_entry *sockfd_cache_grab(const struct node_id *nid,
- int *ret_idx)
-{
- struct sockfd_cache_entry *entry;
-
- pthread_rwlock_rdlock(&sockfd_cache.lock);
- entry = sockfd_cache_search(nid);
- if (!entry) {
- char name[INET6_ADDRSTRLEN];
-
- addr_to_str(name, sizeof(name), nid->addr, 0);
- sd_dprintf("failed node %s:%d", name, nid->port);
- goto out;
- }
-
- *ret_idx = get_free_slot(entry);
- if (*ret_idx == -1)
- entry = NULL;
-out:
- pthread_rwlock_unlock(&sockfd_cache.lock);
- return entry;
-}
-
-static inline bool slots_all_free(struct sockfd_cache_entry *entry)
-{
- int i;
- for (i = 0; i < fds_count; i++)
- if (uatomic_is_true(&entry->fds[i].in_use))
- return false;
- return true;
-}
-
-static inline void destroy_all_slots(struct sockfd_cache_entry *entry)
-{
- int i;
- for (i = 0; i < fds_count; i++)
- if (entry->fds[i].fd != -1)
- close(entry->fds[i].fd);
-}
-
-/*
- * Destroy all the Cached FDs of the node
- *
- * We don't proceed if some other node grab one FD of the node. In this case,
- * the victim node will finally find itself talking to a dead node and call
- * sheep_del_fd() to delete this node from the cache.
- */
-static bool sockfd_cache_destroy(const struct node_id *nid)
-{
- struct sockfd_cache_entry *entry;
-
- pthread_rwlock_wrlock(&sockfd_cache.lock);
- entry = sockfd_cache_search(nid);
- if (!entry) {
- sd_dprintf("It is already destroyed");
- goto false_out;
- }
-
- if (!slots_all_free(entry)) {
- sd_dprintf("Some victim still holds it");
- goto false_out;
- }
-
- rb_erase(&entry->rb, &sockfd_cache.root);
- pthread_rwlock_unlock(&sockfd_cache.lock);
-
- destroy_all_slots(entry);
- free(entry);
-
- return true;
-false_out:
- pthread_rwlock_unlock(&sockfd_cache.lock);
- return false;
-}
-
-/* When node craches, we should delete it from the cache */
-void sockfd_cache_del(const struct node_id *nid)
-{
- char name[INET6_ADDRSTRLEN];
- int n;
-
- if (!sockfd_cache_destroy(nid))
- return;
-
- n = uatomic_sub_return(&sockfd_cache.count, 1);
- addr_to_str(name, sizeof(name), nid->addr, 0);
- sd_dprintf("%s:%d, count %d", name, nid->port, n);
-}
-
-static void sockfd_cache_add_nolock(const struct node_id *nid)
-{
- struct sockfd_cache_entry *new = xmalloc(sizeof(*new));
- int i;
-
- new->fds = xzalloc(sizeof(struct sockfd_cache_fd) * fds_count);
- for (i = 0; i < fds_count; i++)
- new->fds[i].fd = -1;
-
- memcpy(&new->nid, nid, sizeof(struct node_id));
- if (sockfd_cache_insert(new)) {
- free(new);
- return;
- }
- sockfd_cache.count++;
-}
-
-/* Add group of nodes to the cache */
-void sockfd_cache_add_group(const struct sd_node *nodes, int nr)
-{
- const struct sd_node *p;
-
- sd_dprintf("%d", nr);
- pthread_rwlock_wrlock(&sockfd_cache.lock);
- while (nr--) {
- p = nodes + nr;
- sockfd_cache_add_nolock(&p->nid);
- }
- pthread_rwlock_unlock(&sockfd_cache.lock);
-}
-
-/* Add one node to the cache means we can do caching tricks on this node */
-void sockfd_cache_add(const struct node_id *nid)
-{
- struct sockfd_cache_entry *new;
- char name[INET6_ADDRSTRLEN];
- int n, i;
-
- pthread_rwlock_wrlock(&sockfd_cache.lock);
- new = xmalloc(sizeof(*new));
- new->fds = xzalloc(sizeof(struct sockfd_cache_fd) * fds_count);
- for (i = 0; i < fds_count; i++)
- new->fds[i].fd = -1;
-
- memcpy(&new->nid, nid, sizeof(struct node_id));
- if (sockfd_cache_insert(new)) {
- free(new->fds);
- free(new);
- pthread_rwlock_unlock(&sockfd_cache.lock);
- return;
- }
- pthread_rwlock_unlock(&sockfd_cache.lock);
- n = uatomic_add_return(&sockfd_cache.count, 1);
- addr_to_str(name, sizeof(name), nid->addr, 0);
- sd_dprintf("%s:%d, count %d", name, nid->port, n);
-}
-
-static uatomic_bool fds_in_grow;
-static int fds_high_watermark = FDS_WATERMARK(DEFAULT_FDS_COUNT);
-
-static void do_grow_fds(struct work *work)
-{
- struct sockfd_cache_entry *entry;
- struct rb_node *p;
- int old_fds_count, new_fds_count, new_size, i;
-
- sd_dprintf("%d", fds_count);
- pthread_rwlock_wrlock(&sockfd_cache.lock);
- old_fds_count = fds_count;
- new_fds_count = fds_count * 2;
- new_size = sizeof(struct sockfd_cache_fd) * fds_count * 2;
- for (p = rb_first(&sockfd_cache.root); p; p = rb_next(p)) {
- entry = rb_entry(p, struct sockfd_cache_entry, rb);
- entry->fds = xrealloc(entry->fds, new_size);
- for (i = old_fds_count; i < new_fds_count; i++) {
- entry->fds[i].fd = -1;
- uatomic_set_false(&entry->fds[i].in_use);
- }
- }
-
- fds_count *= 2;
- fds_high_watermark = FDS_WATERMARK(fds_count);
- pthread_rwlock_unlock(&sockfd_cache.lock);
-}
-
-static void grow_fds_done(struct work *work)
-{
- sd_dprintf("fd count has been grown into %d", fds_count);
- uatomic_set_false(&fds_in_grow);
- free(work);
-}
-
-static inline void check_idx(int idx)
-{
- struct work *w;
-
- if (idx <= fds_high_watermark)
- return;
- if (!uatomic_set_true(&fds_in_grow))
- return;
-
- w = xmalloc(sizeof(*w));
- w->fn = do_grow_fds;
- w->done = grow_fds_done;
- queue_work(sys->sockfd_wqueue, w);
-}
-
-/* Add the node back if it is still alive */
-static inline int revalidate_node(const struct node_id *nid)
-{
- bool use_io = nid->io_port ? true : false;
- int fd;
-
- if (use_io) {
- fd = connect_to_addr(nid->io_addr, nid->io_port);
- if (fd >= 0)
- goto alive;
- }
- fd = connect_to_addr(nid->addr, nid->port);
- if (fd < 0)
- return false;
-alive:
- close(fd);
- sockfd_cache_add(nid);
- return true;
-}
-
-/* Try to create/get cached IO connection. If failed, fallback to non-IO one */
-static struct sockfd *sockfd_cache_get(const struct node_id *nid)
-{
- struct sockfd_cache_entry *entry;
- struct sockfd *sfd;
- bool use_io = nid->io_port ? true : false;
- const uint8_t *addr = use_io ? nid->io_addr : nid->addr;
- char name[INET6_ADDRSTRLEN];
- int fd, idx, port = use_io ? nid->io_port : nid->port;
-
- addr_to_str(name, sizeof(name), addr, 0);
-grab:
- entry = sockfd_cache_grab(nid, &idx);
- if (!entry) {
- /*
- * The node is deleted, but someone askes us to grab it.
- * The nid is not in the sockfd cache but probably it might be
- * still alive due to broken network connection or was just too
- * busy to serve any request that makes other nodes deleted it
- * from the sockfd cache. In such cases, we need to add it back.
- */
- if (!revalidate_node(nid))
- return NULL;
- goto grab;
- }
- check_idx(idx);
- if (entry->fds[idx].fd != -1) {
- sd_dprintf("%s:%d, idx %d", name, port, idx);
- goto out;
- }
-
- /* Create a new cached connection for this node */
- sd_dprintf("create cache connection %s:%d idx %d", name, port, idx);
- fd = connect_to(name, port);
- if (fd < 0) {
- if (use_io) {
- sd_eprintf("fallback to non-io connection");
- fd = connect_to_addr(nid->addr, nid->port);
- if (fd >= 0)
- goto new;
- }
- uatomic_set_false(&entry->fds[idx].in_use);
- return NULL;
- }
-new:
- entry->fds[idx].fd = fd;
-out:
- sfd = xmalloc(sizeof(*sfd));
- sfd->fd = entry->fds[idx].fd;
- sfd->idx = idx;
- return sfd;
-}
-
-static void sockfd_cache_put(const struct node_id *nid, int idx)
-{
- bool use_io = nid->io_port ? true : false;
- const uint8_t *addr = use_io ? nid->io_addr : nid->addr;
- int port = use_io ? nid->io_port : nid->port;
- struct sockfd_cache_entry *entry;
- char name[INET6_ADDRSTRLEN];
-
- addr_to_str(name, sizeof(name), addr, 0);
- sd_dprintf("%s:%d idx %d", name, port, idx);
-
- pthread_rwlock_rdlock(&sockfd_cache.lock);
- entry = sockfd_cache_search(nid);
- if (entry)
- uatomic_set_false(&entry->fds[idx].in_use);
- pthread_rwlock_unlock(&sockfd_cache.lock);
-}
-
-static void sockfd_cache_close(const struct node_id *nid, int idx)
-{
- bool use_io = nid->io_port ? true : false;
- const uint8_t *addr = use_io ? nid->io_addr : nid->addr;
- int port = use_io ? nid->io_port : nid->port;
- struct sockfd_cache_entry *entry;
- char name[INET6_ADDRSTRLEN];
-
- addr_to_str(name, sizeof(name), addr, 0);
- sd_dprintf("%s:%d idx %d", name, port, idx);
-
- pthread_rwlock_wrlock(&sockfd_cache.lock);
- entry = sockfd_cache_search(nid);
- if (entry) {
- close(entry->fds[idx].fd);
- entry->fds[idx].fd = -1;
- uatomic_set_false(&entry->fds[idx].in_use);
- }
- pthread_rwlock_unlock(&sockfd_cache.lock);
-}
-
-/*
- * Return a sockfd connected to the node to the caller
- *
- * Try to get a 'long' FD as best, which is cached and never closed. If no FD
- * available, we return a 'short' FD which is supposed to be closed by
- * sheep_put_sockfd().
- *
- * ret_idx is opaque to the caller, -1 indicates it is a short FD.
- */
-struct sockfd *sheep_get_sockfd(const struct node_id *nid)
-{
- struct sockfd *sfd;
- int fd;
-
- sfd = sockfd_cache_get(nid);
- if (sfd)
- return sfd;
-
- /* Fallback on a non-io connection that is to be closed shortly */
- fd = connect_to_addr(nid->addr, nid->port);
- if (fd < 0)
- return NULL;
-
- sfd = xmalloc(sizeof(*sfd));
- sfd->idx = -1;
- sfd->fd = fd;
- sd_dprintf("%d", fd);
- return sfd;
-}
-
-/*
- * Release a sockfd connected to the node, which is acquired from
- * sheep_get_sockfd()
- *
- * If it is a long FD, just decrease the refcount to make it available again.
- * If it is a short FD, close it.
- *
- * sheep_put_sockfd() or sheep_del_sockfd() should be paired with
- * sheep_get_sockfd()
- */
-
-void sheep_put_sockfd(const struct node_id *nid, struct sockfd *sfd)
-{
- if (sfd->idx == -1) {
- sd_dprintf("%d", sfd->fd);
- close(sfd->fd);
- free(sfd);
- return;
- }
-
- sockfd_cache_put(nid, sfd->idx);
- free(sfd);
-}
-
-/*
- * Delete a sockfd connected to the node, when node is crashed.
- *
- * If it is a long FD, de-refcount it and tres to destroy all the cached FDs of
- * this node in the cache.
- * If it is a short FD, just close it.
- */
-void sheep_del_sockfd(const struct node_id *nid, struct sockfd *sfd)
-{
- if (sfd->idx == -1) {
- sd_dprintf("%d", sfd->fd);
- close(sfd->fd);
- free(sfd);
- return;
- }
-
- sockfd_cache_close(nid, sfd->idx);
- sockfd_cache_del(nid);
- free(sfd);
-}
-
int sheep_exec_req(const struct node_id *nid, struct sd_req *hdr, void *buf)
{
struct sd_rsp *rsp = (struct sd_rsp *)hdr;
@@ -524,7 +21,7 @@ int sheep_exec_req(const struct node_id *nid, struct sd_req *hdr, void *buf)
assert(is_worker_thread());
- sfd = sheep_get_sockfd(nid);
+ sfd = sockfd_cache_get(nid);
if (!sfd)
return SD_RES_NETWORK_ERROR;
@@ -532,14 +29,14 @@ int sheep_exec_req(const struct node_id *nid, struct sd_req *hdr, void *buf)
MAX_RETRY_COUNT);
if (ret) {
sd_dprintf("remote node might have gone away");
- sheep_del_sockfd(nid, sfd);
+ sockfd_cache_del(nid, sfd);
return SD_RES_NETWORK_ERROR;
}
ret = rsp->result;
if (ret != SD_RES_SUCCESS)
sd_eprintf("failed %s", sd_strerror(ret));
- sheep_put_sockfd(nid, sfd);
+ sockfd_cache_put(nid, sfd);
return ret;
}
--
1.7.10.4
More information about the sheepdog
mailing list