Commit 0b9973f

committed

Return stream frame header binary in dispatch chunk callback

This saves a system call by sending the frame header and the chunk header at the same time. References rabbitmq/osiris#192

1 parent a5106c6 commit 0b9973fCopy full SHA for 0b9973f

File tree

2 files changed

+10

-18

lines changed

deps/rabbitmq_stream/src
- rabbit_stream_reader.erl
rabbitmq-components.mk

2 files changed

+10

-18

lines changed

`‎deps/rabbitmq_stream/src/rabbit_stream_reader.erl`

Lines changed: 9 additions & 17 deletions

Original file line number	Diff line number	Diff line change
`@@ -3568,12 +3568,9 @@ subscription_exists(StreamSubscriptions, SubscriptionId) ->`
`3568`	`3568`	`lists:any(fun(Id) -> Id =:= SubscriptionId end, SubscriptionIds).`
`3569`	`3569`
`3570`	`3570`	`send_file_callback(?VERSION_1,`
`3571`		`- Transport,`
`3572`	`3571`	`_Log,`
`3573`	`3572`	`#consumer{configuration =`
`3574`		`- #consumer_configuration{socket = S,`
`3575`		`- subscription_id =`
`3576`		`- SubscriptionId,`
	`3573`	`+ #consumer_configuration{subscription_id = SubId,`
`3577`	`3574`	`counters = Counters}},`
`3578`	`3575`	`Counter) ->`
`3579`	`3576`	`fun(#{chunk_id := FirstOffsetInChunk, num_entries := NumEntries},`
`@@ -3584,19 +3581,16 @@ send_file_callback(?VERSION_1,`
`3584`	`3581`	`?REQUEST:1,`
`3585`	`3582`	`?COMMAND_DELIVER:15,`
`3586`	`3583`	`?VERSION_1:16,`
`3587`		`- SubscriptionId:8/unsigned>>,`
`3588`		`- Transport:send(S, FrameBeginning),`
	`3584`	`+ SubId:8/unsigned>>,`
`3589`	`3585`	`atomics:add(Counter, 1, Size),`
`3590`	`3586`	`increase_messages_consumed(Counters, NumEntries),`
`3591`		`- set_consumer_offset(Counters, FirstOffsetInChunk)`
	`3587`	`+ set_consumer_offset(Counters, FirstOffsetInChunk),`
	`3588`	`+ FrameBeginning`
`3592`	`3589`	`end;`
`3593`	`3590`	`send_file_callback(?VERSION_2,`
`3594`		`- Transport,`
`3595`	`3591`	`Log,`
`3596`	`3592`	`#consumer{configuration =`
`3597`		`- #consumer_configuration{socket = S,`
`3598`		`- subscription_id =`
`3599`		`- SubscriptionId,`
	`3593`	`+ #consumer_configuration{subscription_id = SubId,`
`3600`	`3594`	`counters = Counters}},`
`3601`	`3595`	`Counter) ->`
`3602`	`3596`	`fun(#{chunk_id := FirstOffsetInChunk, num_entries := NumEntries},`
`@@ -3608,12 +3602,12 @@ send_file_callback(?VERSION_2,`
`3608`	`3602`	`?REQUEST:1,`
`3609`	`3603`	`?COMMAND_DELIVER:15,`
`3610`	`3604`	`?VERSION_2:16,`
`3611`		`- SubscriptionId:8/unsigned,`
	`3605`	`+ SubId:8/unsigned,`
`3612`	`3606`	`CommittedChunkId:64>>,`
`3613`		`- Transport:send(S, FrameBeginning),`
`3614`	`3607`	`atomics:add(Counter, 1, Size),`
`3615`	`3608`	`increase_messages_consumed(Counters, NumEntries),`
`3616`		`- set_consumer_offset(Counters, FirstOffsetInChunk)`
	`3609`	`+ set_consumer_offset(Counters, FirstOffsetInChunk),`
	`3610`	`+ FrameBeginning`
`3617`	`3611`	`end.`
`3618`	`3612`
`3619`	`3613`	`send_chunks(DeliverVersion,`
`@@ -3683,9 +3677,7 @@ send_chunks(DeliverVersion,`
`3683`	`3677`	`Retry,`
`3684`	`3678`	`Counter) ->`
`3685`	`3679`	`case osiris_log:send_file(Socket, Log,`
`3686`		`- send_file_callback(DeliverVersion,`
`3687`		`- Transport,`
`3688`		`- Log,`
	`3680`	`+ send_file_callback(DeliverVersion, Log,`
`3689`	`3681`	`Consumer,`
`3690`	`3682`	`Counter))`
`3691`	`3683`	`of`

`‎rabbitmq-components.mk`

Lines changed: 1 addition & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ dep_jose = hex 1.11.10`
`49`	`49`	`dep_khepri = hex 0.17.1`
`50`	`50`	`dep_khepri_mnesia_migration = hex 0.8.0`
`51`	`51`	`dep_meck = hex 1.0.0`
`52`		`-dep_osiris = git https://github.com/rabbitmq/osiris v1.9.0`
	`52`	`+dep_osiris = git https://github.com/rabbitmq/osiris send-file-improvements`
`53`	`53`	`dep_prometheus = hex 5.1.1`
`54`	`54`	`dep_ra = hex 2.17.0`
`55`	`55`	`dep_ranch = hex 2.2.0`

0 commit comments

Comments

(0)

Navigation Menu

Search code, repositories, users, issues, pull requests...

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Commit 0b9973f

File tree

2 files changed

2 files changed

`‎deps/rabbitmq_stream/src/rabbit_stream_reader.erl`

`‎rabbitmq-components.mk`

0 commit comments