1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.channel.epoll;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufAllocator;
20 import io.netty.channel.AddressedEnvelope;
21 import io.netty.channel.ChannelMetadata;
22 import io.netty.channel.ChannelOutboundBuffer;
23 import io.netty.channel.ChannelPipeline;
24 import io.netty.channel.ChannelPromise;
25 import io.netty.channel.DefaultAddressedEnvelope;
26 import io.netty.channel.unix.DomainDatagramChannel;
27 import io.netty.channel.unix.DomainDatagramChannelConfig;
28 import io.netty.channel.unix.DomainDatagramPacket;
29 import io.netty.channel.unix.DomainDatagramSocketAddress;
30 import io.netty.channel.unix.DomainSocketAddress;
31 import io.netty.channel.unix.IovArray;
32 import io.netty.channel.unix.PeerCredentials;
33 import io.netty.channel.unix.UnixChannelUtil;
34 import io.netty.util.CharsetUtil;
35 import io.netty.util.UncheckedBooleanSupplier;
36 import io.netty.util.internal.StringUtil;
37
38 import java.io.IOException;
39 import java.net.SocketAddress;
40 import java.nio.ByteBuffer;
41
42 import static io.netty.channel.epoll.LinuxSocket.newSocketDomainDgram;
43
44 public final class EpollDomainDatagramChannel extends AbstractEpollChannel implements DomainDatagramChannel {
45
46 private static final ChannelMetadata METADATA = new ChannelMetadata(true, 16);
47
48 private static final String EXPECTED_TYPES =
49 " (expected: " +
50 StringUtil.simpleClassName(DomainDatagramPacket.class) + ", " +
51 StringUtil.simpleClassName(AddressedEnvelope.class) + '<' +
52 StringUtil.simpleClassName(ByteBuf.class) + ", " +
53 StringUtil.simpleClassName(DomainSocketAddress.class) + ">, " +
54 StringUtil.simpleClassName(ByteBuf.class) + ')';
55
56 private volatile boolean connected;
57 private volatile DomainSocketAddress local;
58 private volatile DomainSocketAddress remote;
59
60 private final EpollDomainDatagramChannelConfig config;
61
62 public EpollDomainDatagramChannel() {
63 this(newSocketDomainDgram(), false);
64 }
65
66 public EpollDomainDatagramChannel(int fd) {
67 this(new LinuxSocket(fd), true);
68 }
69
70 private EpollDomainDatagramChannel(LinuxSocket socket, boolean active) {
71 super(null, socket, active, EpollIoOps.EPOLLERR);
72 config = new EpollDomainDatagramChannelConfig(this);
73 }
74
75 @Override
76 public EpollDomainDatagramChannelConfig config() {
77 return config;
78 }
79
80 @Override
81 protected void doBind(SocketAddress localAddress) throws Exception {
82 super.doBind(localAddress);
83 local = (DomainSocketAddress) localAddress;
84 active = true;
85 }
86
87 @Override
88 protected void doClose() throws Exception {
89 super.doClose();
90 connected = active = false;
91 local = null;
92 remote = null;
93 }
94
95 @Override
96 protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
97 if (super.doConnect(remoteAddress, localAddress)) {
98 if (localAddress != null) {
99 local = (DomainSocketAddress) localAddress;
100 }
101 remote = (DomainSocketAddress) remoteAddress;
102 connected = true;
103 return true;
104 }
105 return false;
106 }
107
108 @Override
109 protected void doDisconnect() throws Exception {
110 doClose();
111 }
112
113 @Override
114 protected void doWrite(ChannelOutboundBuffer in) throws Exception {
115 int maxMessagesPerWrite = maxMessagesPerWrite();
116 while (maxMessagesPerWrite > 0) {
117 Object msg = in.current();
118 if (msg == null) {
119 break;
120 }
121
122 try {
123 boolean done = false;
124 for (int i = config().getWriteSpinCount(); i > 0; --i) {
125 if (doWriteMessage(msg)) {
126 done = true;
127 break;
128 }
129 }
130
131 if (done) {
132 in.remove();
133 maxMessagesPerWrite--;
134 } else {
135 break;
136 }
137 } catch (IOException e) {
138 maxMessagesPerWrite--;
139
140
141
142
143 in.remove(e);
144 }
145 }
146
147 if (in.isEmpty()) {
148
149 clearFlag(Native.EPOLLOUT);
150 } else {
151
152 setFlag(Native.EPOLLOUT);
153 }
154 }
155
156 private boolean doWriteMessage(Object msg) throws Exception {
157 final ByteBuf data;
158 DomainSocketAddress remoteAddress;
159 if (msg instanceof AddressedEnvelope) {
160 @SuppressWarnings("unchecked")
161 AddressedEnvelope<ByteBuf, DomainSocketAddress> envelope =
162 (AddressedEnvelope<ByteBuf, DomainSocketAddress>) msg;
163 data = envelope.content();
164 remoteAddress = envelope.recipient();
165 } else {
166 data = (ByteBuf) msg;
167 remoteAddress = null;
168 }
169
170 final int dataLen = data.readableBytes();
171 final long writtenBytes;
172 if (data.hasMemoryAddress()) {
173 long memoryAddress = data.memoryAddress();
174 if (remoteAddress == null) {
175 writtenBytes = socket.sendAddress(memoryAddress, data.readerIndex(), data.writerIndex());
176 } else {
177 writtenBytes = socket.sendToAddressDomainSocket(memoryAddress, data.readerIndex(), data.writerIndex(),
178 remoteAddress.path().getBytes(CharsetUtil.UTF_8));
179 }
180 } else if (data.nioBufferCount() > 1) {
181 IovArray array = ((NativeArrays) registration().attachment()).cleanIovArray();
182 array.add(data, data.readerIndex(), data.readableBytes());
183 int cnt = array.count();
184 assert cnt != 0;
185
186 if (remoteAddress == null) {
187 writtenBytes = socket.writevAddresses(array.memoryAddress(0), cnt);
188 } else {
189 writtenBytes = socket.sendToAddressesDomainSocket(array.memoryAddress(0), cnt,
190 remoteAddress.path().getBytes(CharsetUtil.UTF_8));
191 }
192 } else {
193 ByteBuffer nioData = data.internalNioBuffer(data.readerIndex(), data.readableBytes());
194 if (remoteAddress == null) {
195 writtenBytes = socket.send(nioData, nioData.position(), nioData.limit());
196 } else {
197 writtenBytes = socket.sendToDomainSocket(nioData, nioData.position(), nioData.limit(),
198 remoteAddress.path().getBytes(CharsetUtil.UTF_8));
199 }
200 }
201
202
203 return writtenBytes > 0 || dataLen == 0;
204 }
205
206 @Override
207 protected Object filterOutboundMessage(Object msg) {
208 if (msg instanceof DomainDatagramPacket) {
209 DomainDatagramPacket packet = (DomainDatagramPacket) msg;
210 ByteBuf content = packet.content();
211 return UnixChannelUtil.isBufferCopyNeededForWrite(content) ?
212 new DomainDatagramPacket(newDirectBuffer(packet, content), packet.recipient()) : msg;
213 }
214
215 if (msg instanceof ByteBuf) {
216 ByteBuf buf = (ByteBuf) msg;
217 return UnixChannelUtil.isBufferCopyNeededForWrite(buf) ? newDirectBuffer(buf) : buf;
218 }
219
220 if (msg instanceof AddressedEnvelope) {
221 @SuppressWarnings("unchecked")
222 AddressedEnvelope<Object, SocketAddress> e = (AddressedEnvelope<Object, SocketAddress>) msg;
223 if (e.content() instanceof ByteBuf &&
224 (e.recipient() == null || e.recipient() instanceof DomainSocketAddress)) {
225
226 ByteBuf content = (ByteBuf) e.content();
227 return UnixChannelUtil.isBufferCopyNeededForWrite(content) ?
228 new DefaultAddressedEnvelope<>(
229 newDirectBuffer(e, content), (DomainSocketAddress) e.recipient()) : e;
230 }
231 }
232
233 throw new UnsupportedOperationException(
234 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
235 }
236
237 @Override
238 public boolean isActive() {
239 return socket.isOpen() && (config.getActiveOnOpen() && isRegistered() || active);
240 }
241
242 @Override
243 public boolean isConnected() {
244 return connected;
245 }
246
247 @Override
248 public DomainSocketAddress localAddress() {
249 return (DomainSocketAddress) super.localAddress();
250 }
251
252 @Override
253 protected DomainSocketAddress localAddress0() {
254 return local;
255 }
256
257 @Override
258 public ChannelMetadata metadata() {
259 return METADATA;
260 }
261
262 @Override
263 protected AbstractEpollUnsafe newUnsafe() {
264 return new EpollDomainDatagramChannelUnsafe();
265 }
266
267 @Override
268 protected void doRegister(ChannelPromise promise) {
269 super.doRegister(promise);
270 promise.addListener(f -> {
271 if (f.isSuccess() && isRegistered()) {
272
273
274 submitCurrentOps();
275 }
276 });
277 }
278
279
280
281
282
283 public PeerCredentials peerCredentials() throws IOException {
284 return socket.getPeerCredentials();
285 }
286
287 @Override
288 public DomainSocketAddress remoteAddress() {
289 return (DomainSocketAddress) super.remoteAddress();
290 }
291
292 @Override
293 protected DomainSocketAddress remoteAddress0() {
294 return remote;
295 }
296
297 final class EpollDomainDatagramChannelUnsafe extends AbstractEpollUnsafe {
298
299 @Override
300 void epollInReady() {
301 assert eventLoop().inEventLoop();
302 final DomainDatagramChannelConfig config = config();
303 if (shouldBreakEpollInReady(config)) {
304 clearEpollIn0();
305 return;
306 }
307 final EpollRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
308 final ChannelPipeline pipeline = pipeline();
309 final ByteBufAllocator allocator = config.getAllocator();
310 allocHandle.reset(config);
311
312 Throwable exception = null;
313 try {
314 ByteBuf byteBuf = null;
315 try {
316 boolean connected = isConnected();
317 do {
318 byteBuf = allocHandle.allocate(allocator);
319 allocHandle.attemptedBytesRead(byteBuf.writableBytes());
320
321 final DomainDatagramPacket packet;
322 if (connected) {
323 allocHandle.lastBytesRead(doReadBytes(byteBuf));
324 if (allocHandle.lastBytesRead() <= 0) {
325
326 byteBuf.release();
327 break;
328 }
329 packet = new DomainDatagramPacket(byteBuf, (DomainSocketAddress) localAddress(),
330 (DomainSocketAddress) remoteAddress());
331 } else {
332 final DomainDatagramSocketAddress remoteAddress;
333 if (byteBuf.hasMemoryAddress()) {
334
335 remoteAddress = socket.recvFromAddressDomainSocket(byteBuf.memoryAddress(),
336 byteBuf.writerIndex(), byteBuf.capacity());
337 } else {
338 ByteBuffer nioData = byteBuf.internalNioBuffer(
339 byteBuf.writerIndex(), byteBuf.writableBytes());
340 remoteAddress =
341 socket.recvFromDomainSocket(nioData, nioData.position(), nioData.limit());
342 }
343
344 if (remoteAddress == null) {
345 allocHandle.lastBytesRead(-1);
346 byteBuf.release();
347 break;
348 }
349 DomainSocketAddress localAddress = remoteAddress.localAddress();
350 if (localAddress == null) {
351 localAddress = (DomainSocketAddress) localAddress();
352 }
353 allocHandle.lastBytesRead(remoteAddress.receivedAmount());
354 byteBuf.writerIndex(byteBuf.writerIndex() + allocHandle.lastBytesRead());
355
356 packet = new DomainDatagramPacket(byteBuf, localAddress, remoteAddress);
357 }
358
359 allocHandle.incMessagesRead(1);
360
361 readPending = false;
362 pipeline.fireChannelRead(packet);
363
364 byteBuf = null;
365
366
367
368 } while (allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER));
369 } catch (Throwable t) {
370 if (byteBuf != null) {
371 byteBuf.release();
372 }
373 exception = t;
374 }
375
376 allocHandle.readComplete();
377 pipeline.fireChannelReadComplete();
378
379 if (exception != null) {
380 pipeline.fireExceptionCaught(exception);
381 }
382 } finally {
383 if (shouldStopReading(config)) {
384 clearEpollIn();
385 }
386 }
387 }
388 }
389 }