| 1024 | } |
| 1025 | |
| 1026 | void commitMessages(TLogData* self, |
| 1027 | Reference<LogData> logData, |
| 1028 | Version version, |
| 1029 | const std::vector<TagsAndMessage>& taggedMessages) { |
| 1030 | // SOMEDAY: This method of copying messages is reasonably memory efficient, but it's still a lot of bytes copied. |
| 1031 | // Find a way to do the memory allocation right as we receive the messages in the network layer. |
| 1032 | |
| 1033 | int64_t addedBytes = 0; |
| 1034 | int64_t overheadBytes = 0; |
| 1035 | int expectedBytes = 0; |
| 1036 | int txsBytes = 0; |
| 1037 | |
| 1038 | if (!taggedMessages.size()) { |
| 1039 | return; |
| 1040 | } |
| 1041 | |
| 1042 | int msgSize = 0; |
| 1043 | for (auto& i : taggedMessages) { |
| 1044 | msgSize += i.message.size(); |
| 1045 | } |
| 1046 | |
| 1047 | // Grab the last block in the blocks list so we can share its arena |
| 1048 | // We pop all of the elements of it to create a "fresh" vector that starts at the end of the previous vector |
| 1049 | Standalone<VectorRef<uint8_t>> block; |
| 1050 | if (logData->messageBlocks.empty()) { |
| 1051 | block = Standalone<VectorRef<uint8_t>>(); |
| 1052 | block.reserve(block.arena(), std::max<int64_t>(SERVER_KNOBS->TLOG_MESSAGE_BLOCK_BYTES, msgSize)); |
| 1053 | } else { |
| 1054 | block = logData->messageBlocks.back().second; |
| 1055 | } |
| 1056 | |
| 1057 | block.pop_front(block.size()); |
| 1058 | |
| 1059 | for (auto& msg : taggedMessages) { |
| 1060 | if (msg.message.size() > block.capacity() - block.size()) { |
| 1061 | logData->messageBlocks.emplace_back(version, block); |
| 1062 | addedBytes += int64_t(block.size()) * SERVER_KNOBS->TLOG_MESSAGE_BLOCK_OVERHEAD_FACTOR; |
| 1063 | block = Standalone<VectorRef<uint8_t>>(); |
| 1064 | block.reserve(block.arena(), std::max<int64_t>(SERVER_KNOBS->TLOG_MESSAGE_BLOCK_BYTES, msgSize)); |
| 1065 | } |
| 1066 | |
| 1067 | block.append(block.arena(), msg.message.begin(), msg.message.size()); |
| 1068 | for (auto tag : msg.tags) { |
| 1069 | if (logData->locality == tagLocalitySatellite) { |
| 1070 | if (!(tag.locality == tagLocalityTxs || tag.locality == tagLocalityLogRouter || tag == txsTag)) { |
| 1071 | continue; |
| 1072 | } |
| 1073 | } else if (!(logData->locality == tagLocalitySpecial || logData->locality == tag.locality || |
| 1074 | tag.locality < 0)) { |
| 1075 | continue; |
| 1076 | } |
| 1077 | |
| 1078 | if (tag.locality == tagLocalityLogRouter) { |
| 1079 | if (!logData->logRouterTags) { |
| 1080 | continue; |
| 1081 | } |
| 1082 | tag.id = tag.id % logData->logRouterTags; |
| 1083 | } |
no test coverage detected