1 /* 2 * Copyright 2001-2005, Haiku. 3 * Distributed under the terms of the MIT License. 4 * 5 * Authors: 6 * Pahtz <pahtz@yahoo.com.au> 7 * Axel Dörfler, axeld@pinc-software.de 8 */ 9 10 /** Class for low-overhead port-based messaging */ 11 12 #include <stdio.h> 13 #include <stdlib.h> 14 #include <string.h> 15 #include <new> 16 17 #include <ServerProtocol.h> 18 #include <LinkSender.h> 19 20 #include "link_message.h" 21 22 23 //#define DEBUG_BPORTLINK 24 #ifdef DEBUG_BPORTLINK 25 # include <stdio.h> 26 # define STRACE(x) printf x 27 #else 28 # define STRACE(x) ; 29 #endif 30 31 static const size_t kMaxStringSize = 4096; 32 static const size_t kWatermark = kInitialBufferSize - 24; 33 // if a message is started after this mark, the buffer is flushed automatically 34 35 namespace BPrivate { 36 37 LinkSender::LinkSender(port_id port) 38 : 39 fPort(port), 40 fBuffer(NULL), 41 fBufferSize(0), 42 43 fCurrentEnd(0), 44 fCurrentStart(0), 45 fCurrentStatus(B_OK) 46 { 47 } 48 49 50 LinkSender::~LinkSender() 51 { 52 free(fBuffer); 53 } 54 55 56 void 57 LinkSender::SetPort(port_id port) 58 { 59 fPort = port; 60 } 61 62 63 status_t 64 LinkSender::StartMessage(int32 code, size_t minSize) 65 { 66 // end previous message 67 if (EndMessage() < B_OK) 68 CancelMessage(); 69 70 if (minSize > kMaxBufferSize - sizeof(message_header)) { 71 // we will handle this case in Attach, using an area 72 minSize = sizeof(area_id); 73 } 74 75 minSize += sizeof(message_header); 76 77 // Eventually flush buffer to make space for the new message. 78 // Note, we do not take the actual buffer size into account to not 79 // delay the time between buffer flushes too much. 80 if (fBufferSize > 0 && (minSize > SpaceLeft() || fCurrentStart >= kWatermark)) { 81 status_t status = Flush(); 82 if (status < B_OK) 83 return status; 84 } 85 86 if (minSize > fBufferSize) { 87 if (AdjustBuffer(minSize) != B_OK) 88 return fCurrentStatus = B_NO_MEMORY; 89 } 90 91 message_header *header = (message_header *)(fBuffer + fCurrentStart); 92 header->size = 0; 93 // will be set later 94 header->code = code; 95 header->flags = 0; 96 97 STRACE(("info: LinkSender buffered header %ld (%lx) [%lu %lu %lu].\n", 98 code, code, header->size, header->code, header->flags)); 99 100 fCurrentEnd += sizeof(message_header); 101 return B_OK; 102 } 103 104 105 status_t 106 LinkSender::EndMessage(bool needsReply) 107 { 108 if (fCurrentEnd == fCurrentStart || fCurrentStatus < B_OK) 109 return fCurrentStatus; 110 111 // record the size of the message 112 message_header *header = (message_header *)(fBuffer + fCurrentStart); 113 header->size = CurrentMessageSize(); 114 if (needsReply) 115 header->flags |= needsReply; 116 117 STRACE(("info: LinkSender EndMessage() of size %ld.\n", header->size)); 118 119 // bump to start of next message 120 fCurrentStart = fCurrentEnd; 121 return B_OK; 122 } 123 124 125 void 126 LinkSender::CancelMessage() 127 { 128 fCurrentEnd = fCurrentStart; 129 fCurrentStatus = B_OK; 130 } 131 132 133 status_t 134 LinkSender::Attach(const void *passedData, size_t passedSize) 135 { 136 size_t size = passedSize; 137 const void* data = passedData; 138 139 if (fCurrentStatus < B_OK) 140 return fCurrentStatus; 141 142 if (size == 0) 143 return fCurrentStatus = B_BAD_VALUE; 144 145 if (fCurrentEnd == fCurrentStart) 146 return B_NO_INIT; // need to call StartMessage() first 147 148 bool useArea = false; 149 if (size >= kMaxBufferSize) { 150 useArea = true; 151 size = sizeof(area_id); 152 } 153 154 if (SpaceLeft() < size) { 155 // we have to make space for the data 156 157 status_t status = FlushCompleted(size + CurrentMessageSize()); 158 if (status < B_OK) 159 return fCurrentStatus = status; 160 } 161 162 area_id senderArea = -1; 163 if (useArea) { 164 void* address = NULL; 165 off_t alignedSize = (passedSize + B_PAGE_SIZE) & ~(B_PAGE_SIZE - 1); 166 senderArea = create_area("LinkSenderArea", &address, B_ANY_ADDRESS, 167 alignedSize, B_NO_LOCK, B_READ_AREA | B_WRITE_AREA); 168 169 if (senderArea < B_OK) 170 return senderArea; 171 172 data = &senderArea; 173 memcpy(address, passedData, passedSize); 174 } 175 176 memcpy(fBuffer + fCurrentEnd, data, size); 177 fCurrentEnd += size; 178 179 return B_OK; 180 } 181 182 183 status_t 184 LinkSender::AttachString(const char *string, int32 length) 185 { 186 if (string == NULL) 187 string = ""; 188 189 size_t maxLength = strlen(string); 190 if (length == -1) { 191 length = (int32)maxLength; 192 193 // we should report an error here 194 if (maxLength > kMaxStringSize) 195 length = 0; 196 } else if (length > (int32)maxLength) 197 length = maxLength; 198 199 status_t status = Attach<int32>(length); 200 if (status < B_OK) 201 return status; 202 203 if (length > 0) { 204 status = Attach(string, length); 205 if (status < B_OK) 206 fCurrentEnd -= sizeof(int32); // rewind the transaction 207 } 208 209 return status; 210 } 211 212 213 status_t 214 LinkSender::AdjustBuffer(size_t newSize, char **_oldBuffer) 215 { 216 // make sure the new size is within bounds 217 if (newSize <= kInitialBufferSize) 218 newSize = kInitialBufferSize; 219 else if (newSize > kMaxBufferSize) 220 return B_BUFFER_OVERFLOW; 221 else if (newSize > kInitialBufferSize) 222 newSize = (newSize + B_PAGE_SIZE - 1) & ~(B_PAGE_SIZE - 1); 223 224 if (newSize == fBufferSize) { 225 // keep existing buffer 226 if (_oldBuffer) 227 *_oldBuffer = fBuffer; 228 return B_OK; 229 } 230 231 // create new larger buffer 232 char *buffer = (char *)malloc(newSize); 233 if (buffer == NULL) 234 return B_NO_MEMORY; 235 236 if (_oldBuffer) 237 *_oldBuffer = fBuffer; 238 else 239 free(fBuffer); 240 241 fBuffer = buffer; 242 fBufferSize = newSize; 243 return B_OK; 244 } 245 246 247 status_t 248 LinkSender::FlushCompleted(size_t newBufferSize) 249 { 250 // we need to hide the incomplete message so that it's not flushed 251 int32 end = fCurrentEnd; 252 int32 start = fCurrentStart; 253 fCurrentEnd = fCurrentStart; 254 255 status_t status = Flush(); 256 if (status < B_OK) { 257 fCurrentEnd = end; 258 return status; 259 } 260 261 char *oldBuffer = NULL; 262 status = AdjustBuffer(newBufferSize, &oldBuffer); 263 if (status != B_OK) 264 return status; 265 266 // move the incomplete message to the start of the buffer 267 fCurrentEnd = end - start; 268 if (oldBuffer != fBuffer) { 269 memcpy(fBuffer, oldBuffer + start, fCurrentEnd); 270 free(oldBuffer); 271 } else 272 memmove(fBuffer, fBuffer + start, fCurrentEnd); 273 274 return B_OK; 275 } 276 277 278 status_t 279 LinkSender::Flush(bigtime_t timeout, bool needsReply) 280 { 281 if (fCurrentStatus < B_OK) 282 return fCurrentStatus; 283 284 EndMessage(needsReply); 285 if (fCurrentStart == 0) 286 return B_OK; 287 288 STRACE(("info: LinkSender Flush() waiting to send messages of %ld bytes on port %ld.\n", 289 fCurrentEnd, fPort)); 290 291 status_t err; 292 if (timeout != B_INFINITE_TIMEOUT) { 293 do { 294 err = write_port_etc(fPort, kLinkCode, fBuffer, 295 fCurrentEnd, B_RELATIVE_TIMEOUT, timeout); 296 } while (err == B_INTERRUPTED); 297 } else { 298 do { 299 err = write_port(fPort, kLinkCode, fBuffer, fCurrentEnd); 300 } while (err == B_INTERRUPTED); 301 } 302 303 if (err < B_OK) { 304 STRACE(("error info: LinkSender Flush() failed for %ld bytes (%s) on port %ld.\n", 305 fCurrentEnd, strerror(err), fPort)); 306 return err; 307 } 308 309 STRACE(("info: LinkSender Flush() messages total of %ld bytes on port %ld.\n", 310 fCurrentEnd, fPort)); 311 312 fCurrentEnd = 0; 313 fCurrentStart = 0; 314 315 return B_OK; 316 } 317 318 } // namespace BPrivate 319