b215794de3
Motivation: The writeSpinCount currently loops over the same buffer, gathering write, file write, or other write operation multiple times but will continue writing until there is nothing left or the OS doesn't accept any data for that specific write. However if the OS keeps accepting writes there is no way to limit how much time we spend on a specific socket. This can lead to unfair consumption of resources dedicated to a single socket. We currently don't limit the amount of bytes we attempt to write per gathering write. If there are many more bytes pending relative to the SO_SNDBUF size we will end up building iov arrays with more elements than can be written, which results in extra iteration, conditionals, and book keeping. Modifications: - writeSpinCount should limit the number of system calls we make to write data, instead of applying to individual write operations - IovArray should support a maximum number of bytes - IovArray should support composite buffers of greater than size 1024 - We should auto-scale the amount of data that we attempt to write per gathering write operation relative to SO_SNDBUF and how much data is successfully written - The non-unsafe path should also support a maximum number of bytes, and respect the IOV_MAX limit Result: Write resource consumption can be bounded and gathering writes have a limit relative to the amount of data which can actually be accepted by the socket.
193 lines
6.2 KiB
Java
193 lines
6.2 KiB
Java
/*
|
|
* Copyright 2015 The Netty Project
|
|
*
|
|
* The Netty Project licenses this file to you under the Apache License,
|
|
* version 2.0 (the "License"); you may not use this file except in compliance
|
|
* with the License. You may obtain a copy of the License at:
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
* License for the specific language governing permissions and limitations
|
|
* under the License.
|
|
*/
|
|
package io.netty.channel.epoll;
|
|
|
|
import io.netty.channel.Channel;
|
|
import io.netty.channel.ChannelConfig;
|
|
import io.netty.channel.ChannelOutboundBuffer;
|
|
import io.netty.channel.ChannelPipeline;
|
|
import io.netty.channel.unix.DomainSocketAddress;
|
|
import io.netty.channel.unix.DomainSocketChannel;
|
|
import io.netty.channel.unix.FileDescriptor;
|
|
import io.netty.channel.unix.PeerCredentials;
|
|
import io.netty.util.internal.UnstableApi;
|
|
|
|
import java.io.IOException;
|
|
import java.net.SocketAddress;
|
|
|
|
import static io.netty.channel.epoll.LinuxSocket.newSocketDomain;
|
|
|
|
public final class EpollDomainSocketChannel extends AbstractEpollStreamChannel implements DomainSocketChannel {
|
|
private final EpollDomainSocketChannelConfig config = new EpollDomainSocketChannelConfig(this);
|
|
|
|
private volatile DomainSocketAddress local;
|
|
private volatile DomainSocketAddress remote;
|
|
|
|
public EpollDomainSocketChannel() {
|
|
super(newSocketDomain(), false);
|
|
}
|
|
|
|
EpollDomainSocketChannel(Channel parent, FileDescriptor fd) {
|
|
super(parent, new LinuxSocket(fd.intValue()));
|
|
}
|
|
|
|
public EpollDomainSocketChannel(int fd) {
|
|
super(fd);
|
|
}
|
|
|
|
public EpollDomainSocketChannel(Channel parent, LinuxSocket fd) {
|
|
super(parent, fd);
|
|
}
|
|
|
|
public EpollDomainSocketChannel(int fd, boolean active) {
|
|
super(new LinuxSocket(fd), active);
|
|
}
|
|
|
|
@Override
|
|
protected AbstractEpollUnsafe newUnsafe() {
|
|
return new EpollDomainUnsafe();
|
|
}
|
|
|
|
@Override
|
|
protected DomainSocketAddress localAddress0() {
|
|
return local;
|
|
}
|
|
|
|
@Override
|
|
protected DomainSocketAddress remoteAddress0() {
|
|
return remote;
|
|
}
|
|
|
|
@Override
|
|
protected void doBind(SocketAddress localAddress) throws Exception {
|
|
socket.bind(localAddress);
|
|
local = (DomainSocketAddress) localAddress;
|
|
}
|
|
|
|
@Override
|
|
public EpollDomainSocketChannelConfig config() {
|
|
return config;
|
|
}
|
|
|
|
@Override
|
|
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
|
|
if (super.doConnect(remoteAddress, localAddress)) {
|
|
local = (DomainSocketAddress) localAddress;
|
|
remote = (DomainSocketAddress) remoteAddress;
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
@Override
|
|
public DomainSocketAddress remoteAddress() {
|
|
return (DomainSocketAddress) super.remoteAddress();
|
|
}
|
|
|
|
@Override
|
|
public DomainSocketAddress localAddress() {
|
|
return (DomainSocketAddress) super.localAddress();
|
|
}
|
|
|
|
@Override
|
|
protected int doWriteSingle(ChannelOutboundBuffer in) throws Exception {
|
|
Object msg = in.current();
|
|
if (msg instanceof FileDescriptor && socket.sendFd(((FileDescriptor) msg).intValue()) > 0) {
|
|
// File descriptor was written, so remove it.
|
|
in.remove();
|
|
return 1;
|
|
}
|
|
return super.doWriteSingle(in);
|
|
}
|
|
|
|
@Override
|
|
protected Object filterOutboundMessage(Object msg) {
|
|
if (msg instanceof FileDescriptor) {
|
|
return msg;
|
|
}
|
|
return super.filterOutboundMessage(msg);
|
|
}
|
|
|
|
/**
|
|
* Returns the unix credentials (uid, gid, pid) of the peer
|
|
* <a href=http://man7.org/linux/man-pages/man7/socket.7.html>SO_PEERCRED</a>
|
|
*/
|
|
@UnstableApi
|
|
public PeerCredentials peerCredentials() throws IOException {
|
|
return socket.getPeerCredentials();
|
|
}
|
|
|
|
private final class EpollDomainUnsafe extends EpollStreamUnsafe {
|
|
@Override
|
|
void epollInReady() {
|
|
switch (config().getReadMode()) {
|
|
case BYTES:
|
|
super.epollInReady();
|
|
break;
|
|
case FILE_DESCRIPTORS:
|
|
epollInReadFd();
|
|
break;
|
|
default:
|
|
throw new Error();
|
|
}
|
|
}
|
|
|
|
private void epollInReadFd() {
|
|
if (socket.isInputShutdown()) {
|
|
clearEpollIn0();
|
|
return;
|
|
}
|
|
final ChannelConfig config = config();
|
|
final EpollRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
|
|
allocHandle.edgeTriggered(isFlagSet(Native.EPOLLET));
|
|
|
|
final ChannelPipeline pipeline = pipeline();
|
|
allocHandle.reset(config);
|
|
epollInBefore();
|
|
|
|
try {
|
|
readLoop: do {
|
|
// lastBytesRead represents the fd. We use lastBytesRead because it must be set so that the
|
|
// EpollRecvByteAllocatorHandle knows if it should try to read again or not when autoRead is
|
|
// enabled.
|
|
allocHandle.lastBytesRead(socket.recvFd());
|
|
switch(allocHandle.lastBytesRead()) {
|
|
case 0:
|
|
break readLoop;
|
|
case -1:
|
|
close(voidPromise());
|
|
return;
|
|
default:
|
|
allocHandle.incMessagesRead(1);
|
|
readPending = false;
|
|
pipeline.fireChannelRead(new FileDescriptor(allocHandle.lastBytesRead()));
|
|
break;
|
|
}
|
|
} while (allocHandle.continueReading());
|
|
|
|
allocHandle.readComplete();
|
|
pipeline.fireChannelReadComplete();
|
|
} catch (Throwable t) {
|
|
allocHandle.readComplete();
|
|
pipeline.fireChannelReadComplete();
|
|
pipeline.fireExceptionCaught(t);
|
|
} finally {
|
|
epollInFinally(config);
|
|
}
|
|
}
|
|
}
|
|
}
|