Brpc source code analysis (III) -- the mechanism of requesting other servers and writing data to sockets

brpc As a rpc frame , In addition to supporting as server Of course, you also need to call other server Have good support , This article will focus on brpc Access to other server The specific process and the related past socket Write response The mechanism of .

1.brpc Request other server The way

brpc Sending data in is a call channel class , The key lies in Channel::CallMethod function , Whether it's based on protobuf The orthodox RPC call , still http request , All depend on this function , The definition is as follows ：

The notes explain clearly , There are five parameters

1. method： The remote service to call
2. controller： Contains additional data and various options
3. request： Request to invoke the service
4. response： Return after calling the service
5. done： For callbacks , If the NULL, It will return immediately after sending the request , After the call, it will pass done-Run Execute callback function , It belongs to asynchronous call , If NULL, Will block until the call is completed .
   among 1、3、4 Of or based on protobuf The orthodox rpc Call the required parameters , If it's normal http The request does not require ,

1.1 be based on proto Of rpc Request example

Called Echo The function is pb Generated , as follows ：

1.2 Ordinary http Request example ：

And in the CallMethod Inside , The core is IssueRPC, That is, the part that actually makes calls to the remote end .

stay IssueRPC Inside , First, I took a temporary socket And set the remote address according to the policy of the selected server ,

Then get the appropriate according to the connection type socket, Be careful tmp_socket Use

If it is a single connection, directly put tmp_sock Pass to sending_sock, In case of connection pool or short connection , Based on tmp_socket To get the corresponding connection pool socket Or short connection socket, And release tmp_socket, Pay attention to what you get for sending socket It's all registered epoll in The event , Call after the event is triggered As I said before onNewMessage.

Then it is to package and send data , First, call the corresponding protocol to package ：

call socket Of write send out ：

2.brpc The mechanism of writing data

As mentioned above write In the function , Call StartWrite Start writing , To improve performance and throughput , Sending messages is a little complicated , This is how the official document describes ：

" news ” Point to the binary string with boundary written by the connection , It may be sent upstream client Of response Or downstream server Of request. Multiple threads may send messages to one at the same time fd Send a message , And write fd Non atomic again , So how to queue packets written by different threads efficiently is the key here .baidu-rpc Use a wait-free MPSC Linked list to achieve this function . All the data to be written out is placed in a single linked list node ,next The pointer is initialized to a special value (Socket::WriteRequest::UNCONNECTED). When a thread wants to write data , It first tries to match the corresponding chain header (Socket::_write_head) Do atomic Exchange , The return value is the chain header before exchange . If the return value is empty , It shows that it has obtained the right to write , It will write data once in place . Otherwise, another thread is writing , It is the next The pointer points to the returned header , Then the writing thread will see and write this data . This method can make writing competition be wait-free Of , The thread that obtains the write right is not wait-free Neither lock-free, May be a value of UNCONNECTED Node locking of （ This requires that the thread initiating the write happens after the atomic Exchange , Set up next Before the pointer , Only one instruction is OS Swap out ）, But in practice, it rarely appears . In the current implementation , If the thread that gets the right to write cannot write all the data at once , Will start a KeepWrite The thread continues to write , Until all the data is written . This logic is very complicated , The general principle is as follows , Please read the details socket.cpp.

StartWrite in , If above , The first is to determine whether other threads are writing , If there are other threads writing, they will hang themselves in the linked list and return directly , Because the thread that is writing will continue to write other parts in the linked list after writing its own part . Otherwise, set req->next by NULL, Start writing in place ：

One fd There are three request The schematic diagram of is as follows （ The whole process is the first request Not finished , I've been writing the first request）, Note that in the current state , The linked list is actually reverse ,writehead What is stored in is the recently added

When writing data , First ConnectIfNot, This function has three return values ,-1 0 and 1,0 The logo was originally connected ,-1 Identity connection failed ,1 Means connecting , After connecting, it will execute KeepWriteIfConnected Callback starts writing , It will call KeepWrite：

Pay attention to Connect I can use KeepWriteIfConnected Register as a callback function epoll out event , That is, if there is no connection , Then connect and register epollout,ConnectIfNot return 1,StartWrite Will go straight back to ,epoll out Ensure that once connected, it will continue to write .

If ConnectIfNot return 0, Then keep going ：

Write in place once in the current thread , If it's finished , Before finishing writing, another thread starts writing ：

Iswritecomplete Used to judge whether it is finished , At present request Unfinished and new request It's not finished ：

Function first determines the current request Is it finished ,return_when_no_more There is no new request The return value under , If the current writing is finished, it is true, That is, the whole thing is finished , It is false, Still need to start a new bthread Come on keepwrite.desired Is to give _write_head Value , That is the next part to be written , Initialize to NULL, If the current old_head It's not finished , Then the assignment is old_head. Next call _write_head.compare_exchange_strong, That is, if new_head and _write_head equal , Then put the desired In the write_head, return true, Otherwise, assign new_head Into a _write_head Value , return false.compare_exchange_strong return true Indicates that there is no new req To come over , The new tail is the old head , return return_when_no_more, That is, whether the current node has been written .compare_exchange_strong Failure to return indicates a new request Replaced by write_head in ,new_head Be set as new _write_head The value of the , Keep going down .

As mentioned earlier , In the process of writing, if there is any new req To come over , The linked list is actually reverse , The above figure is the process of overturning ,new_head After the previous process, there is the previous moment _write_head, hypothesis req1 After writing once, I entered Iswritecomplete function , To get the new_head yes 3,old_head yes 1, Then came another 4, So from 3 Start flipping to 1 end , After flipping, it is as follows ,4 There is no sense of existence at this time , This round will not deal with ：

In this case IsWriteComplete return false, therefore StartWrite Function to the next section ：

Write in place , Take the current req Start as a parameter bt perform keepwrite Keep writing ,

keepwrite The main operations are as follows ： Give priority to writing the current req. Only current req Write the next one after writing , Because the previous execution Iswritecomplete Function to flip the known part of the linked list , therefore req->next It's the next one to come req.

Release the completed request：

Find the tail of the current linked list , For example, in the above case, the tail is 3, And use this tail as a parameter to call the one mentioned above iswritecomplete

Here is a detailed explanation KeepWrite call iswritecomplete Parameters of the process , It can be understood in combination with the above linked list diagram ,

KeepWrite Yes, it is cur_tail and req Make parameters to call ,cur_tail As old_head, After calling here, if the tail is updated, the new value will be updated to cur_tail in , Be careful iswritecomplete after , Put aside the new req, The original req They are all overturned , in other words cur_tail It's the new tail .KeepWrite call iswritecomplete It can be divided into four situations , The point is if there are new arrivals req It will trigger the turnover for the subsequent batch ：
1.req == cur_tail , There are new arrivals req
Originally, there was only one to write , But there are newcomers, so compare_exchange_strong Failure , Update the new tail to cur_tail, return false, Not finished
2.req == cur_tail , No newcomers req
Originally, there was only one to write , And there are no newcomers , If this one is not finished, return false, After writing, return true
3.req != cur_tail , There are new arrivals req
because !singular_node establish , therefore return_when_no_more = false, But there are newcomers, so compare_exchange_strong Failure , Do the following flipping operation , For example, the example in the figure above , Call again iswritecomplete after 4 It will become a new tail after turning , Update to cur_tail, Ensure that the following loop is written back to .
4.req != cur_tail , No newcomers req：
because !singular_node establish , therefore return_when_no_more = false, And compare_exchange_strong success , Go straight back to return_when_no_more, That is to say false.

Whole keepwrtie In a do whilie(1) In the structure , That is to say, keep writing before you finish writing , The order is to finish writing the current req Of , Then write the next one of the overturned linked list req, Ensure that the order .

In general, this writing mechanism is more complex , For a trip socket Write the data on , Summarized below ：

1. Atomic operation determines whether other threads are writing this socket 了 , If there is , Connect this write with the original write queue and return directly , What you are writing will be the same after writing your own data socket Write the rest of the data in the queue .
2. If no one writes , Determine whether it is connected , If it is connected, the current thread directly initiates a write to complete the write as soon as possible , Avoid context switching , And if the amount of data is large , I will return immediately after I haven't finished writing once , Start another thread to continue writing , To avoid blocking , Because the current initiated writing is just ordinary threads ,brpc There is no special IO Thread .
3. If you haven't connected yet, start the connection and use “ Continue to write the function ” Register as a callback epollout Back directly , Let the callback complete the subsequent write operation .

The reason for using this complex mechanism , The main purpose is to pursue efficiency 、 Avoid blocking , Increase throughput .