Backpressure questions

[alexolog]

Hello,
I have some questions about backpressure

If I get a DROPPED return code, is it always due to backpressure?
Am I guaranteed to get a .drain event?
I guess an exception would be if the first message is larger than the internal buffer, but in that case I probably cannot send it at all, or is it not the case?
Can I get a DROPPED without getting BACKPRESSURE on a prior send?

[uNetworkingAB]

Yes you get DROPPED because of backpressure but only if your settings (closeOnBackpressure = false, maxBackpressure = some number) require it.

.drain is not necessarily a guarantee, it simply emits whenever the backpressure has reduced or stayed the same. You're not necessarily guaranteed to get .drain for instance if the receiver refuses to ACK the data, you will never drain but you will still gain backpressure.

So treat them as two separate things.

I guess an exception would be if the first message is larger than the internal buffer, but in that case I probably cannot send it at all, or is it not the case?
Can I get a DROPPED without getting BACKPRESSURE on a prior send?

I don't remember but logically it should be "yes". Don't assume anything. DROPPED can come from nowhere.

[alexolog]

Yes you get DROPPED because of backpressure but only if your settings (closeOnBackpressure = false, maxBackpressure = some number) require it.

What happens if maxBackpressure is not explicitly set?

[uNetworkingAB]

Answers to your backpressure questions

Question	Answer
Is DROPPED always due to backpressure?	Yes — in the current send() implementation, DROPPED is only returned when maxBackpressure is exceeded.
Are you guaranteed a .drain event after BACKPRESSURE?	Not strictly guaranteed on every occurrence, but practically yes if data remains buffered — .drain fires from on_writable, and also whenever write()/uncork() reduces backpressure.
Can the first message be too big to send at all?	No such special exception exists — even a large first message goes through the normal buffering path and returns BACKPRESSURE, not a hard failure.
Can you get DROPPED without a prior BACKPRESSURE?	Yes — DROPPED is checked against getBufferedAmount() before any bytes are sent, so if buffered amount is already over maxBackpressure (built up from prior successful/backpressured sends, or from pub/sub broadcasts), a completely fresh send() call can return DROPPED immediately without you personally having received BACKPRESSURE on an immediately preceding call.

---

Details

DROPPED is solely a backpressure-limit signal

WebSocket::send() checks buffered amount against maxBackpressure at the very top, before any actual write attempt 1 . If exceeded, it returns DROPPED and, optionally, closes the read side, plus invokes droppedHandler if set 2 . There is no other code path returning DROPPED in send() — the only two other outcomes are SUCCESS and BACKPRESSURE, so this is the sole trigger.

maxBackpressure is a per-route setting configured in WebSocketBehavior (default 64KB) 3 , and copied into WebSocketContextData when the route is set up 4 .

.drain and its relationship to BACKPRESSURE

.drain is fired from the socket's on_writable handler in WebSocketContext, which is invoked by the underlying uSockets library whenever the kernel signals the socket is writable again 5 . Specifically, it drains as much of the buffered data as possible via write(nullptr, 0), then calls drainHandler only if backpressure actually decreased, or if it started at 0 6 .

This means:

• If you sent and got BACKPRESSURE, .drain will fire the next time the OS reports writable and buffered amount decreases — this is effectively guaranteed as long as the connection stays open and the peer keeps reading.
• If the connection closes/times out before draining completes, .drain will simply never fire again (per docs: "on disconnect all resources are freed, including ... any backpressure") — see the wiki notes on ping/pong heartbeats.

The wiki itself states this nuance directly: "When send returns BACKPRESSURE it means you should stop sending data until the drain event fires...", and ".drain event is only a hint that it has changed" — meaning you shouldn't assume a 1:1 guarantee timing-wise, only that it will eventually fire if backpressure is draining.

No special case for an oversized "first message"

There is a special-cased fast path in send() for large (>= 16KB), uncompressed, non-SSL, non-subscriber messages sent while buffer is empty and uncorked 7 . Even here, if the OS write is incomplete, the leftover is simply appended to webSocketData->buffer and BACKPRESSURE is returned — there's no failure mode for "too big to send" 8 .

For the general path, Super::getSendBuffer() in AsyncSocket always returns some buffer — either the cork buffer, or falls back to resizing the per-socket BackPressure buffer (std::string-like, dynamically growable) to fit the message 9 . So there's no fixed-size internal buffer limit that would reject a first message outright — the only ceiling is maxBackpressure, checked before buffering even happens.

DROPPED without a prior BACKPRESSURE from you

Because the maxBackpressure check in send() looks only at current getBufferedAmount() 10 , backpressure can build up from sources other than your own explicit send() calls returning BACKPRESSURE — e.g., pub/sub broadcast delivery via TopicTree::drain 11 , or a burst where a single call itself already puts you over the limit on the very first send(). In these cases, your very next (or even first) send() call can return DROPPED directly, with no BACKPRESSURE observed by you beforehand.

Notes

• This behavior is documented informally in the project's misc/READMORE.md "Backpressure" section, which matches the code: BACKPRESSURE/SUCCESS/DROPPED semantics, maxBackpressure enforcement, and the .drain hint caveat 12 .
• closeOnBackpressureLimit (default false) can cause the read side to be shut down immediately upon hitting the DROPPED case, which is a related but distinct effect from the drop itself 13 .

Wiki pages you might want to explore:

• Architecture (uNetworking/uWebSockets)
• HTTP to WebSocket Upgrade (uNetworking/uWebSockets)

Citations

File: src/WebSocket.h (L112-132)

    /* Send or buffer a WebSocket frame, compressed or not. Returns BACKPRESSURE on increased user space backpressure,
     * DROPPED on dropped message (due to backpressure) or SUCCCESS if you are free to send even more now. */
    SendStatus send(std::string_view message, OpCode opCode = OpCode::BINARY, int compress = false, bool fin = true) {
        WebSocketContextData<SSL, USERDATA> *webSocketContextData = (WebSocketContextData<SSL, USERDATA> *) us_socket_context_ext(SSL,
            (us_socket_context_t *) us_socket_context(SSL, (us_socket_t *) this)
        );

        /* Skip sending and report success if we are over the limit of maxBackpressure */
        if (webSocketContextData->maxBackpressure && webSocketContextData->maxBackpressure < getBufferedAmount()) {
            /* Also defer a close if we should */
            if (webSocketContextData->closeOnBackpressureLimit) {
                us_socket_shutdown_read(SSL, (us_socket_t *) this);
            }

            /* It is okay to call send again from within this callback since we immediately return with DROPPED afterwards */
            if (webSocketContextData->droppedHandler) {
                webSocketContextData->droppedHandler(this, message, opCode);
            }

            return DROPPED;
        }

File: src/WebSocket.h (L138-156)

        if (message.length() >= 16 * 1024 && !compress && !SSL && !webSocketData->subscriber && getBufferedAmount() == 0 && Super::getLoopData()->corkOffset == 0) {
            char header[10];
            int header_length = (int) protocol::formatMessage<isServer>(header, "", 0, opCode, message.length(), compress, fin);
            int written = us_socket_write2(0, (struct us_socket_t *)this, header, header_length, message.data(), (int) message.length());
        
            if (written != header_length + (int) message.length()) {
                /* Buffer up backpressure */
                if (written > header_length) {
                    webSocketData->buffer.append(message.data() + written - header_length, message.length() - (size_t) (written - header_length));
                } else {
                    webSocketData->buffer.append(header + written, (size_t) header_length - (size_t) written);
                    webSocketData->buffer.append(message.data(), message.length());
                }
                /* We cannot still be corked if we have backpressure.
                 * We also cannot uncork normally since it will re-write the already buffered
                 * up backpressure again. */
                Super::uncorkWithoutSending();
                return BACKPRESSURE;
            }

File: src/WebSocket.h (L159-162)

            if (webSocketData->subscriber) {
                /* This will call back into us, send. */
                webSocketContextData->topicTree->drain(webSocketData->subscriber);
            }

File: src/App.h (L263-265)

        /* 64kb backpressure is probably good */
        unsigned int maxBackpressure = 64 * 1024;
        bool closeOnBackpressureLimit = false;

File: src/App.h (L414-416)

        webSocketContext->getExt()->maxPayloadLength = behavior.maxPayloadLength;
        webSocketContext->getExt()->maxBackpressure = behavior.maxBackpressure;
        webSocketContext->getExt()->closeOnBackpressureLimit = behavior.closeOnBackpressureLimit;

File: src/WebSocketContext.h (L326-372)

        us_socket_context_on_writable(SSL, getSocketContext(), [](auto *s) {

            /* NOTE: Are we called here corked? If so, the below write code is broken, since
             * we will have 0 as getBufferedAmount due to writing to cork buffer, then sending TCP FIN before
             * we actually uncorked and sent off things */

            /* It makes sense to check for us_is_shut_down here and return if so, to avoid shutting down twice */
            if (us_socket_is_shut_down(SSL, (us_socket_t *) s)) {
                return s;
            }

            AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
            WebSocketData *webSocketData = (WebSocketData *)(us_socket_ext(SSL, s));

            /* We store old backpressure since it is unclear whether write drained anything,
             * however, in case of coming here with 0 backpressure we still need to emit drain event */
            unsigned int backpressure = asyncSocket->getBufferedAmount();

            /* Drain as much as possible */
            asyncSocket->write(nullptr, 0);

            /* Behavior: if we actively drain backpressure, always reset timeout (even if we are in shutdown) */
            /* Also reset timeout if we came here with 0 backpressure */
            if (!backpressure || backpressure > asyncSocket->getBufferedAmount()) {
                auto *webSocketContextData = (WebSocketContextData<SSL, USERDATA> *) us_socket_context_ext(SSL, us_socket_context(SSL, (us_socket_t *) s));
                asyncSocket->timeout(webSocketContextData->idleTimeoutComponents.first);
                webSocketData->hasTimedOut = false;
            }

            /* Are we in (WebSocket) shutdown mode? */
            if (webSocketData->isShuttingDown) {
                /* Check if we just now drained completely */
                if (asyncSocket->getBufferedAmount() == 0) {
                    /* Now perform the actual TCP/TLS shutdown which was postponed due to backpressure */
                    asyncSocket->shutdown();
                }
            } else if (!backpressure || backpressure > asyncSocket->getBufferedAmount()) {
                /* Only call drain if we actually drained backpressure or if we came here with 0 backpressure */
                auto *webSocketContextData = (WebSocketContextData<SSL, USERDATA> *) us_socket_context_ext(SSL, us_socket_context(SSL, (us_socket_t *) s));
                if (webSocketContextData->drainHandler) {
                    webSocketContextData->drainHandler((WebSocket<SSL, isServer, USERDATA> *) s);
                }
                /* No need to check for closed here as we leave the handler immediately*/
            }

            return s;
        });

File: src/AsyncSocket.h (L154-189)

    /* Returns a suitable buffer for temporary assemblation of send data */
    std::pair<char *, SendBufferAttribute> getSendBuffer(size_t size) {
        /* First step is to determine if we already have backpressure or not */
        LoopData *loopData = getLoopData();
        BackPressure &backPressure = getAsyncSocketData()->buffer;
        size_t existingBackpressure = backPressure.length();
        if ((!existingBackpressure) && (isCorked() || canCork()) && (loopData->corkOffset + size < LoopData::CORK_BUFFER_SIZE)) {
            /* Cork automatically if we can */
            if (isCorked()) {
                char *sendBuffer = loopData->corkBuffer + loopData->corkOffset;
                loopData->corkOffset += (unsigned int) size;
                return {sendBuffer, SendBufferAttribute::NEEDS_NOTHING};
            } else {
                cork();
                char *sendBuffer = loopData->corkBuffer + loopData->corkOffset;
                loopData->corkOffset += (unsigned int) size;
                return {sendBuffer, SendBufferAttribute::NEEDS_UNCORK};
            }
        } else {

            /* If we are corked and there is already data in the cork buffer,
            mark how much is ours and reset it */
            unsigned int ourCorkOffset = 0;
            if (isCorked() && loopData->corkOffset) {
                ourCorkOffset = loopData->corkOffset;
                loopData->corkOffset = 0;
            }

            /* Fallback is to use the backpressure as buffer */
            backPressure.resize(ourCorkOffset + existingBackpressure + size);

            /* And copy corkbuffer in front */
            memcpy((char *) backPressure.data() + existingBackpressure, loopData->corkBuffer, ourCorkOffset);

            return {(char *) backPressure.data() + ourCorkOffset + existingBackpressure, SendBufferAttribute::NEEDS_DRAIN};
        }

File: misc/READMORE.md (L175-176)

#### Backpressure
Sending on a WebSocket can build backpressure. WebSocket::send returns an enum of BACKPRESSURE, SUCCESS or DROPPED. When send returns BACKPRESSURE it means you should stop sending data until the drain event fires and WebSocket::getBufferedAmount() returns a reasonable amount of bytes. But in case you specified a maxBackpressure when creating the WebSocketContext, this limit will automatically be enforced. That means an attempt at sending a message which would result in too much backpressure will be canceled and send will return DROPPED. This means the message was dropped and will not be put in the queue. maxBackpressure is an essential setting when using pub/sub as a slow receiver otherwise could build up a lot of backpressure. By setting maxBackpressure the library will automatically mana ... (truncated)

[alexolog]

Thank you!