Utility functions for reference counting.

adsbus/exec.c

@@ -32,7 +32,8 @@ static struct list_head exec_head = LIST_HEAD_INIT(exec_head);
 static void exec_spawn_wrapper(struct peer *);
 
 static void exec_del(struct exec *exec) {
-	(*exec->flow->ref_count)--;
+	flow_ref_dec(exec->flow);
+
 	if (exec->child > 0) {
 		fprintf(stderr, "E %s: Sending SIGTERM to child process %d\n", exec->id, exec->child);
 		// Racy with the process terminating, so don't assert on it
@@ -121,7 +122,7 @@ void exec_cleanup() {
 }
 
 void exec_new(char *command, struct flow *flow, void *passthrough) {
-	(*flow->ref_count)++;
+	flow_ref_inc(flow);
 
 	struct exec *exec = malloc(sizeof(*exec));
 	assert(exec);

adsbus/file.c

@@ -59,7 +59,7 @@ static bool file_should_retry(int fd, struct file *file) {
 }
 
 static void file_del(struct file *file) {
-	(*file->flow->ref_count)--;
+	flow_ref_dec(file->flow);
 	list_del(&file->file_list);
 	free(file->path);
 	free(file);
@@ -109,7 +109,7 @@ static void file_open_wrapper(struct peer *peer) {
 }
 
 static void file_new(char *path, int flags, struct flow *flow, void *passthrough) {
-	(*flow->ref_count)++;
+	flow_ref_inc(flow);
 
 	struct file *file = malloc(sizeof(*file));
 	assert(file);

adsbus/flow.c

@@ -31,3 +31,11 @@ bool flow_new(int fd, struct flow *flow, void *passthrough) {
 	flow->new(fd, passthrough, NULL);
 	return true;
 }
+
+void flow_ref_inc(struct flow *flow) {
+	(*flow->ref_count)++;
+}
+
+void flow_ref_dec(struct flow *flow) {
+	(*flow->ref_count)--;
+}

adsbus/flow.h

@@ -17,3 +17,5 @@ struct flow {
 void flow_socket_connected(int, struct flow *);
 bool flow_hello(int, struct flow *, void *);
 bool flow_new(int, struct flow *, void *);
+void flow_ref_inc(struct flow *);
+void flow_ref_dec(struct flow *);

adsbus/incoming.c

@@ -74,7 +74,7 @@ static void incoming_handler(struct peer *peer) {
 }
 
 static void incoming_del(struct incoming *incoming) {
-	(*incoming->flow->ref_count)--;
+	flow_ref_dec(incoming->flow);
 	if (incoming->peer.fd >= 0) {
 		assert(!close(incoming->peer.fd));
 	}
@@ -151,7 +151,7 @@ void incoming_cleanup() {
 }
 
 void incoming_new(char *node, char *service, struct flow *flow, void *passthrough) {
-	(*flow->ref_count)++;
+	flow_ref_inc(flow);
 
 	struct incoming *incoming = malloc(sizeof(*incoming));
 	incoming->peer.event_handler = incoming_handler;

adsbus/outgoing.c

@@ -142,7 +142,7 @@ static void outgoing_resolve_wrapper(struct peer *peer) {
 }
 
 static void outgoing_del(struct outgoing *outgoing) {
-	(*outgoing->flow->ref_count)--;
+	flow_ref_dec(outgoing->flow);
 	if (outgoing->peer.fd >= 0) {
 		assert(!close(outgoing->peer.fd));
 	}
@@ -160,7 +160,7 @@ void outgoing_cleanup() {
 }
 
 void outgoing_new(char *node, char *service, struct flow *flow, void *passthrough) {
-	(*flow->ref_count)++;
+	flow_ref_inc(flow);
 
 	struct outgoing *outgoing = malloc(sizeof(*outgoing));
 	uuid_gen(outgoing->id);