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
27 import java.io.IOException;
28 import java.net.InetSocketAddress;
29 import java.net.SocketAddress;
30 import static io.netty.channel.unix.Errors.ioResult;
31
32 public final class IoUringSocketChannel extends AbstractIoUringStreamChannel implements SocketChannel {
33 private final IoUringSocketChannelConfig config;
34
35 public IoUringSocketChannel() {
36 super(null, LinuxSocket.newSocketStream(), false);
37 this.config = new IoUringSocketChannelConfig(this);
38 }
39
40 IoUringSocketChannel(Channel parent, LinuxSocket fd) {
41 super(parent, fd, true);
42 this.config = new IoUringSocketChannelConfig(this);
43 }
44
45 IoUringSocketChannel(Channel parent, LinuxSocket fd, SocketAddress remote) {
46 super(parent, fd, remote);
47 this.config = new IoUringSocketChannelConfig(this);
48 }
49
50
51
52
53
54 public IoUringTcpInfo tcpInfo() {
55 return tcpInfo(new IoUringTcpInfo());
56 }
57
58
59
60
61
62 public IoUringTcpInfo tcpInfo(IoUringTcpInfo info) {
63 try {
64 socket.getTcpInfo(info);
65 return info;
66 } catch (IOException e) {
67 throw new ChannelException(e);
68 }
69 }
70
71 @Override
72 public ServerSocketChannel parent() {
73 return (ServerSocketChannel) super.parent();
74 }
75
76 @Override
77 public SocketChannelConfig config() {
78 return config;
79 }
80
81 @Override
82 public InetSocketAddress remoteAddress() {
83 return (InetSocketAddress) super.remoteAddress();
84 }
85
86 @Override
87 public InetSocketAddress localAddress() {
88 return (InetSocketAddress) super.localAddress();
89 }
90
91 @Override
92 protected AbstractUringUnsafe newUnsafe() {
93 return new IoUringSocketUnsafe();
94 }
95
96 private final class IoUringSocketUnsafe extends IoUringStreamUnsafe {
97 @Override
98 protected int scheduleWriteSingle(Object msg) {
99 assert writeId == 0;
100
101 if (IoUring.isSendZcSupported() && msg instanceof ByteBuf) {
102 ByteBuf buf = (ByteBuf) msg;
103 int length = buf.readableBytes();
104 if (((IoUringSocketChannelConfig) config()).shouldWriteZeroCopy(length)) {
105 long address = IoUring.memoryAddress(buf) + buf.readerIndex();
106 long opsId = writeTracker.nextZeroCopyId();
107 IoUringIoOps ops = IoUringIoOps.newSendZc(fd().intValue(), address, length, 0, opsId, 0);
108 byte opCode = ops.opcode();
109 writeTracker.record(opsId, opCode, buf);
110 writeId = registration().submit(ops);
111 writeOpCode = opCode;
112 if (writeId == 0) {
113 writeTracker.abandon(opsId, opCode);
114 return 0;
115 }
116 return 1;
117 }
118
119 }
120 return super.scheduleWriteSingle(msg);
121 }
122
123 @Override
124 protected int scheduleWriteMultiple(ChannelOutboundBuffer in) {
125 assert writeId == 0;
126
127 IoUringSocketChannelConfig ioUringSocketChannelConfig = (IoUringSocketChannelConfig) config();
128
129 if (IoUring.isSendmsgZcSupported()
130 && (ioUringSocketChannelConfig.shouldWriteZeroCopy(((ByteBuf) in.current()).readableBytes()))) {
131 IoUringIoHandler handler = registration().attachment();
132
133 IovArray iovArray = handler.iovArray();
134 int offset = iovArray.count();
135 IovArrayReferenceCollector collector = handler.iovArrayReferenceCollector();
136 try {
137
138
139 iovArray.maxCount(Native.MAX_SKB_FRAGS);
140 try {
141 in.forEachFlushedMessage(new ChannelOutboundBuffer.MessageProcessor() {
142 @Override
143 public boolean processMessage(Object msg) throws Exception {
144 if (msg instanceof ByteBuf) {
145 ByteBuf buf = (ByteBuf) msg;
146 int length = buf.readableBytes();
147 if (ioUringSocketChannelConfig.shouldWriteZeroCopy(length)) {
148 return collector.processMessage(msg);
149 }
150 }
151 return false;
152 }
153 });
154 } catch (Exception e) {
155
156 return scheduleWriteSingle(in.current());
157 }
158 long iovArrayAddress = iovArray.memoryAddress(offset);
159 int iovArrayLength = iovArray.count() - offset;
160
161 MsgHdrMemoryArray msgHdrArray = handler.msgHdrMemoryArray();
162 MsgHdrMemory hdr = msgHdrArray.nextHdr();
163 assert hdr != null;
164 hdr.set(iovArrayAddress, iovArrayLength);
165 long opsId = writeTracker.nextZeroCopyId();
166 IoUringIoOps ops = IoUringIoOps.newSendmsgZc(
167 fd().intValue(), (byte) 0, 0, hdr.address(), opsId);
168 byte opCode = ops.opcode();
169 writeTracker.record(opsId, opCode, collector.referencesArray(), collector.referencesCount());
170 writeId = registration().submit(ops);
171 writeOpCode = opCode;
172 if (writeId == 0) {
173 writeTracker.abandon(opsId, opCode);
174 return 0;
175 }
176 return 1;
177 } finally {
178
179
180 collector.reset();
181 }
182 }
183
184 return super.scheduleWriteMultiple(in);
185 }
186
187 @Override
188 protected ChannelOutboundBuffer.MessageProcessor filterWriteMultiple(IovArrayReferenceCollector collector) {
189 if (!IoUring.isSendmsgZcSupported()) {
190 return super.filterWriteMultiple(collector);
191 }
192 IoUringSocketChannelConfig ioUringSocketChannelConfig = (IoUringSocketChannelConfig) config();
193 return new ChannelOutboundBuffer.MessageProcessor() {
194 @Override
195 public boolean processMessage(Object msg) throws Exception {
196 if (msg instanceof ByteBuf) {
197 ByteBuf buf = (ByteBuf) msg;
198 int length = buf.readableBytes();
199 if (ioUringSocketChannelConfig.shouldWriteZeroCopy(length)) {
200 return false;
201 }
202 }
203 return collector.processMessage(msg);
204 }
205 };
206 }
207
208 @Override
209 boolean writeComplete0(byte op, int res, int flags, long data, int outstanding) {
210 if (op == Native.IORING_OP_SEND_ZC || op == Native.IORING_OP_SENDMSG_ZC) {
211 return handleWriteCompleteZeroCopy(op, res, flags, data);
212 }
213 return super.writeComplete0(op, res, flags, data, outstanding);
214 }
215
216 private boolean handleWriteCompleteZeroCopy(byte op, int res, int flags, long data) {
217 if ((flags & Native.IORING_CQE_F_NOTIF) != 0) {
218 return true;
219 }
220 writeId = 0;
221 writeOpCode = 0;
222 if ((flags & Native.IORING_CQE_F_MORE) != 0) {
223
224
225
226 writeTracker.retainReferences(data, op);
227 }
228 ChannelOutboundBuffer channelOutboundBuffer = outboundBuffer();
229 if (channelOutboundBuffer == null) {
230 return true;
231 }
232 if (res >= 0) {
233 channelOutboundBuffer.removeBytes(res);
234 return true;
235 }
236 if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
237 return true;
238 }
239 try {
240 return ioResult(op == Native.IORING_OP_SEND_ZC ? "io_uring sendzc" : "io_uring sendmsg_zc", res) != 0;
241 } catch (Throwable cause) {
242 handleWriteError(cause);
243 return true;
244 }
245 }
246 }
247 }