1 /* 2 * Authored by Alex Hultman, 2018-2020. 3 * Intellectual property of third-party. 4 5 * Licensed under the Apache License, Version 2.0 (the "License"); 6 * you may not use this file except in compliance with the License. 7 * You may obtain a copy of the License at 8 9 * http://www.apache.org/licenses/LICENSE-2.0 10 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 18 #ifndef UWS_ASYNCSOCKET_H 19 #define UWS_ASYNCSOCKET_H 20 21 /* This class implements async socket memory management strategies */ 22 23 /* NOTE: Many unsigned/signed conversion warnings could be solved by moving from int length 24 * to unsigned length for everything to/from uSockets - this would however remove the opportunity 25 * to signal error with -1 (which is how the entire UNIX syscalling is built). */ 26 27 #include <cstring> 28 #include <iostream> 29 30 #include "libusockets.h" 31 32 #include "LoopData.h" 33 #include "AsyncSocketData.h" 34 35 namespace uWS { 36 37 enum SendBufferAttribute { 38 NEEDS_NOTHING, 39 NEEDS_DRAIN, 40 NEEDS_UNCORK 41 }; 42 43 template <bool, bool, typename> struct WebSocketContext; 44 45 template <bool SSL> 46 struct AsyncSocket { 47 /* This guy is promiscuous */ 48 template <bool> friend struct HttpContext; 49 template <bool, bool, typename> friend struct WebSocketContext; 50 template <bool> friend struct TemplatedApp; 51 template <bool, typename> friend struct WebSocketContextData; 52 template <typename, typename> friend struct TopicTree; 53 54 protected: 55 /* Returns SSL pointer or FD as pointer */ getNativeHandleAsyncSocket56 void *getNativeHandle() { 57 return us_socket_get_native_handle(SSL, (us_socket_t *) this); 58 } 59 60 /* Get loop data for socket */ getLoopDataAsyncSocket61 LoopData *getLoopData() { 62 return (LoopData *) us_loop_ext(us_socket_context_loop(SSL, us_socket_context(SSL, (us_socket_t *) this))); 63 } 64 65 /* Get socket extension */ getAsyncSocketDataAsyncSocket66 AsyncSocketData<SSL> *getAsyncSocketData() { 67 return (AsyncSocketData<SSL> *) us_socket_ext(SSL, (us_socket_t *) this); 68 } 69 70 /* Socket timeout */ timeoutAsyncSocket71 void timeout(unsigned int seconds) { 72 us_socket_timeout(SSL, (us_socket_t *) this, seconds); 73 } 74 75 /* Shutdown socket without any automatic drainage */ shutdownAsyncSocket76 void shutdown() { 77 us_socket_shutdown(SSL, (us_socket_t *) this); 78 } 79 80 /* Immediately close socket */ closeAsyncSocket81 us_socket_t *close() { 82 return us_socket_close(SSL, (us_socket_t *) this, 0, nullptr); 83 } 84 corkUncheckedAsyncSocket85 void corkUnchecked() { 86 /* What if another socket is corked? */ 87 getLoopData()->corkedSocket = this; 88 } 89 90 /* Cork this socket. Only one socket may ever be corked per-loop at any given time */ corkAsyncSocket91 void cork() { 92 /* Extra check for invalid corking of others */ 93 if (getLoopData()->corkOffset && getLoopData()->corkedSocket != this) { 94 std::cerr << "Error: Cork buffer must not be acquired without checking canCork!" << std::endl; 95 std::terminate(); 96 } 97 98 /* What if another socket is corked? */ 99 getLoopData()->corkedSocket = this; 100 } 101 102 /* Returns wheter we are corked or not */ isCorkedAsyncSocket103 bool isCorked() { 104 return getLoopData()->corkedSocket == this; 105 } 106 107 /* Returns whether we could cork (it is free) */ canCorkAsyncSocket108 bool canCork() { 109 return getLoopData()->corkedSocket == nullptr; 110 } 111 112 /* Returns a suitable buffer for temporary assemblation of send data */ getSendBufferAsyncSocket113 std::pair<char *, SendBufferAttribute> getSendBuffer(size_t size) { 114 /* First step is to determine if we already have backpressure or not */ 115 LoopData *loopData = getLoopData(); 116 BackPressure &backPressure = getAsyncSocketData()->buffer; 117 size_t existingBackpressure = backPressure.length(); 118 if ((!existingBackpressure) && (isCorked() || canCork()) && (loopData->corkOffset + size < LoopData::CORK_BUFFER_SIZE)) { 119 /* Cork automatically if we can */ 120 if (isCorked()) { 121 char *sendBuffer = loopData->corkBuffer + loopData->corkOffset; 122 loopData->corkOffset += (unsigned int) size; 123 return {sendBuffer, SendBufferAttribute::NEEDS_NOTHING}; 124 } else { 125 cork(); 126 char *sendBuffer = loopData->corkBuffer + loopData->corkOffset; 127 loopData->corkOffset += (unsigned int) size; 128 return {sendBuffer, SendBufferAttribute::NEEDS_UNCORK}; 129 } 130 } else { 131 132 /* If we are corked and there is already data in the cork buffer, 133 mark how much is ours and reset it */ 134 unsigned int ourCorkOffset = 0; 135 if (isCorked() && loopData->corkOffset) { 136 ourCorkOffset = loopData->corkOffset; 137 loopData->corkOffset = 0; 138 } 139 140 /* Fallback is to use the backpressure as buffer */ 141 backPressure.resize(ourCorkOffset + existingBackpressure + size); 142 143 /* And copy corkbuffer in front */ 144 memcpy((char *) backPressure.data() + existingBackpressure, loopData->corkBuffer, ourCorkOffset); 145 146 return {(char *) backPressure.data() + ourCorkOffset + existingBackpressure, SendBufferAttribute::NEEDS_DRAIN}; 147 } 148 } 149 150 /* Returns the user space backpressure. */ getBufferedAmountAsyncSocket151 unsigned int getBufferedAmount() { 152 /* We return the actual amount of bytes in backbuffer, including pendingRemoval */ 153 return (unsigned int) getAsyncSocketData()->buffer.totalLength(); 154 } 155 156 /* Returns the text representation of an IPv4 or IPv6 address */ addressAsTextAsyncSocket157 std::string_view addressAsText(std::string_view binary) { 158 static thread_local char buf[64]; 159 int ipLength = 0; 160 161 if (!binary.length()) { 162 return {}; 163 } 164 165 unsigned char *b = (unsigned char *) binary.data(); 166 167 if (binary.length() == 4) { 168 ipLength = sprintf(buf, "%u.%u.%u.%u", b[0], b[1], b[2], b[3]); 169 } else { 170 ipLength = sprintf(buf, "%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x:%02x%02x", 171 b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8], b[9], b[10], b[11], 172 b[12], b[13], b[14], b[15]); 173 } 174 175 return {buf, (unsigned int) ipLength}; 176 } 177 178 /* Returns the remote IP address or empty string on failure */ getRemoteAddressAsyncSocket179 std::string_view getRemoteAddress() { 180 static thread_local char buf[16]; 181 int ipLength = 16; 182 us_socket_remote_address(SSL, (us_socket_t *) this, buf, &ipLength); 183 return std::string_view(buf, (unsigned int) ipLength); 184 } 185 186 /* Returns the text representation of IP */ getRemoteAddressAsTextAsyncSocket187 std::string_view getRemoteAddressAsText() { 188 return addressAsText(getRemoteAddress()); 189 } 190 191 /* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. 192 * Returns pair of bytes written (anywhere) and wheter or not this call resulted in the polling for 193 * writable (or we are in a state that implies polling for writable). */ 194 std::pair<int, bool> write(const char *src, int length, bool optionally = false, int nextLength = 0) { 195 /* Fake success if closed, simple fix to allow uncork of closed socket to succeed */ 196 if (us_socket_is_closed(SSL, (us_socket_t *) this)) { 197 return {length, false}; 198 } 199 200 LoopData *loopData = getLoopData(); 201 AsyncSocketData<SSL> *asyncSocketData = getAsyncSocketData(); 202 203 /* We are limited if we have a per-socket buffer */ 204 if (asyncSocketData->buffer.length()) { 205 /* Write off as much as we can */ 206 int written = us_socket_write(SSL, (us_socket_t *) this, asyncSocketData->buffer.data(), (int) asyncSocketData->buffer.length(), /*nextLength != 0 | */length); 207 208 /* On failure return, otherwise continue down the function */ 209 if ((unsigned int) written < asyncSocketData->buffer.length()) { 210 211 /* Update buffering (todo: we can do better here if we keep track of what happens to this guy later on) */ 212 asyncSocketData->buffer.erase((unsigned int) written); 213 214 if (optionally) { 215 /* Thankfully we can exit early here */ 216 return {0, true}; 217 } else { 218 /* This path is horrible and points towards erroneous usage */ 219 asyncSocketData->buffer.append(src, (unsigned int) length); 220 221 return {length, true}; 222 } 223 } 224 225 /* At this point we simply have no buffer and can continue as normal */ 226 asyncSocketData->buffer.clear(); 227 } 228 229 if (length) { 230 if (loopData->corkedSocket == this) { 231 /* We are corked */ 232 if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= (unsigned int) length) { 233 /* If the entire chunk fits in cork buffer */ 234 memcpy(loopData->corkBuffer + loopData->corkOffset, src, (unsigned int) length); 235 loopData->corkOffset += (unsigned int) length; 236 /* Fall through to default return */ 237 } else { 238 /* Strategy differences between SSL and non-SSL regarding syscall minimizing */ 239 if constexpr (SSL) { 240 /* Cork up as much as we can */ 241 unsigned int stripped = LoopData::CORK_BUFFER_SIZE - loopData->corkOffset; 242 memcpy(loopData->corkBuffer + loopData->corkOffset, src, stripped); 243 loopData->corkOffset = LoopData::CORK_BUFFER_SIZE; 244 245 auto [written, failed] = uncork(src + stripped, length - (int) stripped, optionally); 246 return {written + (int) stripped, failed}; 247 } 248 249 /* For non-SSL we take the penalty of two syscalls */ 250 return uncork(src, length, optionally); 251 } 252 } else { 253 /* We are not corked */ 254 int written = us_socket_write(SSL, (us_socket_t *) this, src, length, nextLength != 0); 255 256 /* Did we fail? */ 257 if (written < length) { 258 /* If the write was optional then just bail out */ 259 if (optionally) { 260 return {written, true}; 261 } 262 263 /* Fall back to worst possible case (should be very rare for HTTP) */ 264 /* At least we can reserve room for next chunk if we know it up front */ 265 if (nextLength) { 266 asyncSocketData->buffer.reserve(asyncSocketData->buffer.length() + (size_t) (length - written + nextLength)); 267 } 268 269 /* Buffer this chunk */ 270 asyncSocketData->buffer.append(src + written, (size_t) (length - written)); 271 272 /* Return the failure */ 273 return {length, true}; 274 } 275 /* Fall through to default return */ 276 } 277 } 278 279 /* Default fall through return */ 280 return {length, false}; 281 } 282 283 /* Uncork this socket and flush or buffer any corked and/or passed data. It is essential to remember doing this. */ 284 /* It does NOT count bytes written from cork buffer (they are already accounted for in the write call responsible for its corking)! */ 285 std::pair<int, bool> uncork(const char *src = nullptr, int length = 0, bool optionally = false) { 286 LoopData *loopData = getLoopData(); 287 288 if (loopData->corkedSocket == this) { 289 loopData->corkedSocket = nullptr; 290 291 if (loopData->corkOffset) { 292 /* Corked data is already accounted for via its write call */ 293 auto [written, failed] = write(loopData->corkBuffer, (int) loopData->corkOffset, false, length); 294 loopData->corkOffset = 0; 295 296 if (failed) { 297 /* We do not need to care for buffering here, write does that */ 298 return {0, true}; 299 } 300 } 301 302 /* We should only return with new writes, not things written to cork already */ 303 return write(src, length, optionally, 0); 304 } else { 305 /* We are not even corked! */ 306 return {0, false}; 307 } 308 } 309 }; 310 311 } 312 313 #endif // UWS_ASYNCSOCKET_H 314