Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

exchange_server supports dynamically modifying the fullOp of the ringbus in sock_recv #883

Merged
merged 3 commits into from
Mar 3, 2024
Merged

exchange_server supports dynamically modifying the fullOp of the ringbus in sock_recv #883

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8adde09
* ADD [ringbuffer] New Ringbuffer api for change fullop
RanMaoyi Mar 1, 2024
47de9d1
* MDF [exchange] tss[2] = 2 means change ringbuffer operation when qu…
RanMaoyi Mar 1, 2024
89f0c2d
* MDF [ringbuffer] Rename ringBuffer_fullOp to ringBuffer_set_fullOp
RanMaoyi Mar 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions include/nng/supplemental/nanolib/ringbuffer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,8 @@ int ringBuffer_search_msgs_fuzz(ringBuffer_t *rb,
nng_msg ***list);
int ringBuffer_get_and_clean_msgs(ringBuffer_t *rb,
unsigned int *count, nng_msg ***list);

int ringBuffer_set_fullOp(ringBuffer_t *rb, enum fullOption fullOp);
#ifdef SUPP_PARQUET
int ringBuffer_get_msgs_from_file(ringBuffer_t *rb, void ***msgs, int **msgLen);
int ringBuffer_get_msgs_from_file_by_keys(ringBuffer_t *rb, uint64_t *keys, uint32_t count,
Expand Down
15 changes: 14 additions & 1 deletion src/mqtt/protocol/exchange/exchange_server.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,7 +342,7 @@ exchange_sock_recv(void *arg, nni_aio *aio)
nni_aio_finish_error(aio, NNG_EINVAL);
return;
}
} else {
} else if (tss[2] == 1) {
/* clean up and return */
/* Only one exchange with one ringBuffer now */
nng_mtx_lock(s->ex_node->ex->rbs[0]->ring_lock);
Expand All @@ -355,6 +355,19 @@ exchange_sock_recv(void *arg, nni_aio *aio)
return;
}
nng_mtx_unlock(s->ex_node->ex->rbs[0]->ring_lock);
} else if (tss[2] == 2) {
/* Change MQ fullOp to tss[1] */
/* Only one exchange with one ringBuffer now */
nng_mtx_lock(s->ex_node->ex->rbs[0]->ring_lock);
ret = ringBuffer_set_fullOp(s->ex_node->ex->rbs[0], tss[1]);
if (ret != 0) {
log_warn("ringBuffer_fullOp failed!");
nng_mtx_unlock(s->ex_node->ex->rbs[0]->ring_lock);
nni_mtx_unlock(&s->mtx);
nni_aio_finish_error(aio, NNG_EINVAL);
return;
}
nng_mtx_unlock(s->ex_node->ex->rbs[0]->ring_lock);
}
}

Expand Down
12 changes: 12 additions & 0 deletions src/supplemental/nanolib/ringbuffer/ringbuffer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -695,6 +695,18 @@ static int put_msgs_to_aio(ringBuffer_t *rb, nng_aio *aio)
return 0;
}

int ringBuffer_set_fullOp(ringBuffer_t *rb, enum fullOption fullOp)
{
if (rb == NULL || fullOp >= RB_FULL_MAX) {
log_error("ringbuffer is NULL or fullOp is not valid: %d\n", fullOp);
return -1;
}

rb->fullOp = fullOp;

return 0;
}

int ringBuffer_enqueue(ringBuffer_t *rb,
uint64_t key,
void *data,
Expand Down
Loading