Skip to content

Commit

Permalink
fixes #1408 Reference count as a first class type
Browse files Browse the repository at this point in the history
This starts by using this for the nni_pipe, but we will use it
for the other primary objects as well.  This should simplify
the tear down and hopefully eliminate some races.

It does mean that pipe destruction goes through an additional
context switch, for now at least.  This shouldn't be on the hot
data path anyway.
  • Loading branch information
gdamore committed Dec 7, 2024
1 parent 513f9d1 commit 755dcb2
Show file tree
Hide file tree
Showing 6 changed files with 89 additions and 31 deletions.
2 changes: 2 additions & 0 deletions src/core/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@ nng_sources(
protocol.h
reap.c
reap.h
refcnt.c
refcnt.h
sockaddr.c
socket.c
socket.h
Expand Down
1 change: 1 addition & 0 deletions src/core/nng_impl.h
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
#include "core/pollable.h"
#include "core/protocol.h"
#include "core/reap.h"
#include "core/refcnt.h"
#include "core/stats.h"
#include "core/stream.h"
#include "core/strs.h"
Expand Down
53 changes: 24 additions & 29 deletions src/core/pipe.c
Original file line number Diff line number Diff line change
Expand Up @@ -25,19 +25,31 @@ static nni_id_map pipes =
static nni_mtx pipes_lk = NNI_MTX_INITIALIZER;

static void pipe_destroy(void *);
static void pipe_reap(void *);

static nni_reap_list pipe_reap_list = {
.rl_offset = offsetof(nni_pipe, p_reap),
.rl_func = pipe_destroy,
.rl_func = pipe_reap,
};

static void
pipe_destroy(void *arg)
{
nni_pipe *p = arg;
if (p == NULL) {
return;

if (p->p_proto_data != NULL) {
p->p_proto_ops.pipe_fini(p->p_proto_data);
}
if (p->p_tran_data != NULL) {
p->p_tran_ops.p_fini(p->p_tran_data);
}
nni_free(p, p->p_size);
}

void
pipe_reap(void *arg)
{
nni_pipe *p = arg;

nni_pipe_run_cb(p, NNG_PIPE_EV_REM_POST);

Expand All @@ -47,32 +59,21 @@ pipe_destroy(void *arg)
if (p->p_id != 0) {
nni_id_remove(&pipes, p->p_id);
}
// This wait guarantees that all callers are done with us.
while (p->p_ref != 0) {
nni_cv_wait(&p->p_cv);
}
nni_mtx_unlock(&pipes_lk);

if (p->p_proto_data != NULL) {
p->p_proto_ops.pipe_stop(p->p_proto_data);
}
if ((p->p_tran_data != NULL) && (p->p_tran_ops.p_stop != NULL)) {
p->p_tran_ops.p_stop(p->p_tran_data);
}

#ifdef NNG_ENABLE_STATS
nni_stat_unregister(&p->st_root);
#endif
nni_pipe_remove(p);

if (p->p_proto_data != NULL) {
p->p_proto_ops.pipe_fini(p->p_proto_data);
p->p_proto_ops.pipe_stop(p->p_proto_data);
}
if (p->p_tran_data != NULL) {
p->p_tran_ops.p_fini(p->p_tran_data);
if ((p->p_tran_data != NULL) && (p->p_tran_ops.p_stop != NULL)) {
p->p_tran_ops.p_stop(p->p_tran_data);
}
nni_cv_fini(&p->p_cv);
nni_free(p, p->p_size);

nni_pipe_rele(p);
}

int
Expand All @@ -86,7 +87,7 @@ nni_pipe_find(nni_pipe **pp, uint32_t id)
// close the pipe.
nni_mtx_lock(&pipes_lk);
if ((p = nni_id_get(&pipes, id)) != NULL) {
p->p_ref++;
nni_refcnt_hold(&p->p_refcnt);
*pp = p;
}
nni_mtx_unlock(&pipes_lk);
Expand All @@ -96,12 +97,7 @@ nni_pipe_find(nni_pipe **pp, uint32_t id)
void
nni_pipe_rele(nni_pipe *p)
{
nni_mtx_lock(&pipes_lk);
p->p_ref--;
if (p->p_ref == 0) {
nni_cv_wake(&p->p_cv);
}
nni_mtx_unlock(&pipes_lk);
nni_refcnt_rele(&p->p_refcnt);
}

// nni_pipe_id returns the 32-bit pipe id, which can be used in backtraces.
Expand Down Expand Up @@ -250,15 +246,14 @@ pipe_create(nni_pipe **pp, nni_sock *sock, nni_sp_tran *tran, void *tran_data)
p->p_proto_ops = *pops;
p->p_sock = sock;
p->p_cbs = false;
p->p_ref = 1;

nni_refcnt_init(&p->p_refcnt, 2, p, pipe_destroy);

nni_atomic_init_bool(&p->p_closed);
nni_atomic_flag_reset(&p->p_stop);
NNI_LIST_NODE_INIT(&p->p_sock_node);
NNI_LIST_NODE_INIT(&p->p_ep_node);

nni_cv_init(&p->p_cv, &pipes_lk);

nni_mtx_lock(&pipes_lk);
rv = nni_id_alloc32(&pipes, &p->p_id, p);
nni_mtx_unlock(&pipes_lk);
Expand Down
33 changes: 33 additions & 0 deletions src/core/refcnt.c
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
// Copyright 2024 Staysail Systems, Inc. <[email protected]>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt). A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#include <core/refcnt.h>

void
nni_refcnt_init(
nni_refcnt *rc, unsigned value, void *data, void (*fini)(void *))
{
nni_atomic_init(&rc->rc_cnt);
nni_atomic_set(&rc->rc_cnt, value);
rc->rc_data = data;
rc->rc_fini = fini;
}

void
nni_refcnt_hold(nni_refcnt *rc)
{
nni_atomic_inc(&rc->rc_cnt);
}

void
nni_refcnt_rele(nni_refcnt *rc)
{
if (nni_atomic_dec_nv(&rc->rc_cnt) == 0) {
rc->rc_fini(rc->rc_data);
}
}
28 changes: 28 additions & 0 deletions src/core/refcnt.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
// Copyright 2024 Staysail Systems, Inc. <[email protected]>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt). A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#ifndef CORE_REFCNT_H
#define CORE_REFCNT_H

#include <nng/nng.h>

#include <core/nng_impl.h>
#include <core/platform.h>

typedef struct {
nni_atomic_int rc_cnt;
void (*rc_fini)(void *);
void *rc_data;
} nni_refcnt;

extern void nni_refcnt_init(
nni_refcnt *rc, unsigned value, void *v, void (*fini)(void *));
extern void nni_refcnt_hold(nni_refcnt *rc);
extern void nni_refcnt_rele(nni_refcnt *rc);

#endif // CORE_REFCNT_H
3 changes: 1 addition & 2 deletions src/core/sockimpl.h
Original file line number Diff line number Diff line change
Expand Up @@ -105,9 +105,8 @@ struct nni_pipe {
nni_atomic_bool p_closed;
nni_atomic_flag p_stop;
bool p_cbs;
int p_ref;
nni_cv p_cv;
nni_reap_node p_reap;
nni_refcnt p_refcnt;

#ifdef NNG_ENABLE_STATS
nni_stat_item st_root;
Expand Down

0 comments on commit 755dcb2

Please sign in to comment.