1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.channel.uring;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.channel.Channel;
20 import io.netty.channel.ChannelException;
21 import io.netty.channel.ChannelOutboundBuffer;
22 import io.netty.channel.socket.ServerSocketChannel;
23 import io.netty.channel.socket.SocketChannel;
24 import io.netty.channel.socket.SocketChannelConfig;
25 import io.netty.channel.unix.IovArray;
26 import io.netty.util.ReferenceCounted;
27
28 import java.io.IOException;
29 import java.net.InetSocketAddress;
30 import java.net.SocketAddress;
31 import java.util.Collections;
32 import java.util.List;
33
34 import static io.netty.channel.unix.Errors.ioResult;
35
36 public final class IoUringSocketChannel extends AbstractIoUringStreamChannel implements SocketChannel {
37 private final IoUringSocketChannelConfig config;
38
39 public IoUringSocketChannel() {
40 super(null, LinuxSocket.newSocketStream(), false);
41 this.config = new IoUringSocketChannelConfig(this);
42 }
43
44 IoUringSocketChannel(Channel parent, LinuxSocket fd) {
45 super(parent, fd, true);
46 this.config = new IoUringSocketChannelConfig(this);
47 }
48
49 IoUringSocketChannel(Channel parent, LinuxSocket fd, SocketAddress remote) {
50 super(parent, fd, remote);
51 this.config = new IoUringSocketChannelConfig(this);
52 }
53
54
55
56
57
58 public IoUringTcpInfo tcpInfo() {
59 return tcpInfo(new IoUringTcpInfo());
60 }
61
62
63
64
65
66 public IoUringTcpInfo tcpInfo(IoUringTcpInfo info) {
67 try {
68 socket.getTcpInfo(info);
69 return info;
70 } catch (IOException e) {
71 throw new ChannelException(e);
72 }
73 }
74
75 @Override
76 public ServerSocketChannel parent() {
77 return (ServerSocketChannel) super.parent();
78 }
79
80 @Override
81 public SocketChannelConfig config() {
82 return config;
83 }
84
85 @Override
86 public InetSocketAddress remoteAddress() {
87 return (InetSocketAddress) super.remoteAddress();
88 }
89
90 @Override
91 public InetSocketAddress localAddress() {
92 return (InetSocketAddress) super.localAddress();
93 }
94
95 @Override
96 protected AbstractUringUnsafe newUnsafe() {
97 return new IoUringSocketUnsafe();
98 }
99
100 private final class IoUringSocketUnsafe extends IoUringStreamUnsafe {
101
102
103
104 private PendingZeroCopyWrites pendingZeroCopyWrites;
105
106 @Override
107 protected int scheduleWriteSingle(Object msg) {
108 assert writeId == 0;
109
110 if (IoUring.isSendZcSupported() && msg instanceof ByteBuf) {
111 ByteBuf buf = (ByteBuf) msg;
112 int length = buf.readableBytes();
113 if (((IoUringSocketChannelConfig) config()).shouldWriteZeroCopy(length)) {
114 long address = IoUring.memoryAddress(buf) + buf.readerIndex();
115 long opsId = nextZeroCopyUserData();
116 IoUringIoOps ops = IoUringIoOps.newSendZc(fd().intValue(), address, length, 0, opsId, 0);
117 byte opCode = ops.opcode();
118 writeId = submitWrite(ops);
119 writeOpCode = opCode;
120 if (writeId == 0) {
121 pendingZeroCopyWrites.recycle(opsId);
122 return 0;
123 }
124 return 1;
125 }
126
127 }
128 return super.scheduleWriteSingle(msg);
129 }
130
131 @Override
132 protected int scheduleWriteMultiple(ChannelOutboundBuffer in) {
133 assert writeId == 0;
134
135 IoUringSocketChannelConfig ioUringSocketChannelConfig = (IoUringSocketChannelConfig) config();
136
137 if (IoUring.isSendmsgZcSupported()
138 && (ioUringSocketChannelConfig.shouldWriteZeroCopy(((ByteBuf) in.current()).readableBytes()))) {
139 IoUringIoHandler handler = registration().attachment();
140
141 IovArray iovArray = handler.iovArray();
142 int offset = iovArray.count();
143 try {
144 in.forEachFlushedMessage(new ChannelOutboundBuffer.MessageProcessor() {
145 @Override
146 public boolean processMessage(Object msg) throws Exception {
147 if (msg instanceof ByteBuf) {
148 ByteBuf buf = (ByteBuf) msg;
149 int length = buf.readableBytes();
150 if (ioUringSocketChannelConfig.shouldWriteZeroCopy(length)) {
151 return iovArray.processMessage(msg);
152 }
153 }
154 return false;
155 }
156 });
157 } catch (Exception e) {
158
159 return scheduleWriteSingle(in.current());
160 }
161 long iovArrayAddress = iovArray.memoryAddress(offset);
162 int iovArrayLength = iovArray.count() - offset;
163
164 MsgHdrMemory hdr = writeMsgHdr();
165 hdr.set(iovArrayAddress, iovArrayLength);
166 long opsId = nextZeroCopyUserData();
167 IoUringIoOps ops = IoUringIoOps.newSendmsgZc(
168 fd().intValue(), (byte) 0, 0, hdr.address(), opsId);
169 byte opCode = ops.opcode();
170 writeId = submitWrite(ops);
171 writeOpCode = opCode;
172 if (writeId == 0) {
173 pendingZeroCopyWrites.recycle(opsId);
174 return 0;
175 }
176 return 1;
177 }
178
179 return super.scheduleWriteMultiple(in);
180 }
181
182 @Override
183 protected ChannelOutboundBuffer.MessageProcessor filterWriteMultiple(IovArray iovArray) {
184 if (!IoUring.isSendmsgZcSupported()) {
185 return super.filterWriteMultiple(iovArray);
186 }
187 IoUringSocketChannelConfig ioUringSocketChannelConfig = (IoUringSocketChannelConfig) config();
188 return new ChannelOutboundBuffer.MessageProcessor() {
189 @Override
190 public boolean processMessage(Object msg) throws Exception {
191 if (msg instanceof ByteBuf) {
192 ByteBuf buf = (ByteBuf) msg;
193 int length = buf.readableBytes();
194 if (ioUringSocketChannelConfig.shouldWriteZeroCopy(length)) {
195 return false;
196 }
197 }
198 return iovArray.processMessage(msg);
199 }
200 };
201 }
202
203 private long nextZeroCopyUserData() {
204 PendingZeroCopyWrites pendingWrites = pendingZeroCopyWrites;
205 if (pendingWrites == null) {
206 pendingWrites = new PendingZeroCopyWrites();
207 pendingZeroCopyWrites = pendingWrites;
208 }
209 return pendingWrites.nextUserData();
210 }
211
212 @Override
213 List<ReferenceCounted> retainWriteBuffers(ChannelOutboundBuffer buffer) {
214 if (currentWrite.opcode() == Native.IORING_OP_SENDMSG_ZC) {
215 return new IovBufferCollector().collect(buffer, writeMsgHdr().iovLength());
216 }
217 return super.retainWriteBuffers(buffer);
218 }
219
220 @Override
221 boolean writeComplete0(byte op, int res, int flags, long data, int outstanding) {
222 if (op == Native.IORING_OP_SEND_ZC || op == Native.IORING_OP_SENDMSG_ZC) {
223 return handleWriteCompleteZeroCopy(op, res, flags, data);
224 }
225 return super.writeComplete0(op, res, flags, data, outstanding);
226 }
227
228 private void takeSendZCBuffers(long data) {
229 PendingZeroCopyWrites.PendingWrite write = pendingZeroCopyWrites.register(data);
230 if (retainedWriteBuffers == null) {
231 if (currentWrite.opcode() == Native.IORING_OP_SENDMSG_ZC) {
232 write.retain(outboundBuffer(), writeMsgHdr().iovLength());
233 } else {
234 write.add(((ByteBuf) outboundBuffer().current()).retain());
235 }
236 } else {
237
238 for (int i = 0; i < retainedWriteBuffers.size(); i++) {
239 write.add(retainedWriteBuffers.get(i));
240 }
241 retainedWriteBuffers = Collections.emptyList();
242 }
243 }
244
245 private boolean handleWriteCompleteZeroCopy(byte op, int res, int flags, long data) {
246 if ((flags & Native.IORING_CQE_F_NOTIF) != 0) {
247 pendingZeroCopyWrites.release(data);
248 return true;
249 }
250 writeId = 0;
251 writeOpCode = 0;
252 if ((flags & Native.IORING_CQE_F_MORE) != 0) {
253
254 takeSendZCBuffers(data);
255 } else {
256 pendingZeroCopyWrites.recycle(data);
257 }
258 ChannelOutboundBuffer channelOutboundBuffer = outboundBuffer();
259 if (channelOutboundBuffer == null) {
260 return true;
261 }
262 if (res >= 0) {
263 channelOutboundBuffer.removeBytes(res);
264 return true;
265 }
266 if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
267 return true;
268 }
269 try {
270 return ioResult(op == Native.IORING_OP_SEND_ZC ? "io_uring sendzc" : "io_uring sendmsg_zc", res) != 0;
271 } catch (Throwable cause) {
272 handleWriteError(cause);
273 return true;
274 }
275 }
276 }
277 }