1 // Copyright 2021 gRPC authors.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 #include "test/core/promise/benchmark/filter_stack.h"
16
17 namespace filter_stack {
18
MakeChannel(Filter ** filters,size_t num_filters)19 ChannelStack* MakeChannel(Filter** filters, size_t num_filters) {
20 size_t size = sizeof(ChannelStack) + num_filters * sizeof(ChannelElem);
21 size_t call_size = sizeof(CallStack) + num_filters * sizeof(CallElem);
22 for (size_t i = 0; i < num_filters; i++) {
23 size += filters[i]->sizeof_channel_data;
24 call_size += filters[i]->sizeof_call_data;
25 }
26 char* data = new char[size];
27 ChannelStack* stk = reinterpret_cast<ChannelStack*>(data);
28 new (data) ChannelStack{0, num_filters, call_size};
29 data += sizeof(ChannelStack);
30 char* user_data = data + num_filters * sizeof(ChannelElem);
31 for (size_t i = 0; i < num_filters; i++) {
32 new (data) ChannelElem{filters[i], user_data};
33 filters[i]->init_channel_data(reinterpret_cast<ChannelElem*>(data));
34 data += sizeof(ChannelElem);
35 user_data += filters[i]->sizeof_channel_data;
36 }
37 printf("CALL STACK SIZE: %d\n", static_cast<int>(call_size));
38 return stk;
39 }
40
FreeChannel(ChannelStack * stk)41 void FreeChannel(ChannelStack* stk) {
42 ChannelElem* elems = reinterpret_cast<ChannelElem*>(stk + 1);
43 for (size_t i = 0; i < stk->num_elems; i++) {
44 elems[i].filter->destroy_channel_data(&elems[i]);
45 }
46 stk->~ChannelStack();
47 delete[] reinterpret_cast<char*>(stk);
48 }
49
MakeCall(ChannelStack * stk)50 CallStack* MakeCall(ChannelStack* stk) {
51 char* data = new char[stk->call_stack_size];
52 CallStack* call = reinterpret_cast<CallStack*>(data);
53 new (data) CallStack{{1}, stk->num_elems, {}};
54 data += sizeof(CallStack);
55 ChannelElem* channel_elems = reinterpret_cast<ChannelElem*>(stk + 1);
56 char* user_data = data + stk->num_elems * sizeof(CallElem);
57 for (size_t i = 0; i < stk->num_elems; i++) {
58 new (data) CallElem{channel_elems[i].filter, channel_elems[i].channel_data,
59 user_data};
60 channel_elems[i].filter->init_call_data(reinterpret_cast<CallElem*>(data));
61 data += sizeof(CallElem);
62 user_data += channel_elems[i].filter->sizeof_call_data;
63 }
64 return call;
65 }
66
RefCall(CallStack * stk)67 static void RefCall(CallStack* stk) {
68 stk->refcount.fetch_add(1, std::memory_order_relaxed);
69 }
70
UnrefCall(CallStack * stk)71 static void UnrefCall(CallStack* stk) {
72 if (stk->refcount.fetch_sub(1, std::memory_order_acq_rel) == 1) {
73 CallElem* elems = reinterpret_cast<CallElem*>(stk + 1);
74 for (size_t i = 0; i < stk->num_elems; i++) {
75 elems[i].filter->destroy_call_data(&elems[i]);
76 }
77 stk->~CallStack();
78 delete[] reinterpret_cast<char*>(stk);
79 }
80 }
81
FreeCall(CallStack * stk)82 void FreeCall(CallStack* stk) { UnrefCall(stk); }
83
NoChannelData(ChannelElem *)84 void NoChannelData(ChannelElem*) {}
NoCallData(CallElem *)85 void NoCallData(CallElem*) {}
86
StartOp(CallElem * elem,Op * op)87 static void StartOp(CallElem* elem, Op* op) {
88 elem->filter->start_transport_stream_op_batch(elem, op);
89 }
90
CallNextOp(CallElem * elem,Op * op)91 void CallNextOp(CallElem* elem, Op* op) { StartOp(elem + 1, op); }
92
RunOp(CallStack * stk,Op * op)93 void RunOp(CallStack* stk, Op* op) {
94 RefCall(stk);
95 {
96 absl::MutexLock lock(&stk->mutex);
97 StartOp(reinterpret_cast<CallElem*>(stk + 1), op);
98 }
99 UnrefCall(stk);
100 }
101
102 } // namespace filter_stack
103