001: /*
002: * Copyright (c) 2000 by Matt Welsh and The Regents of the University of
003: * California. All rights reserved.
004: *
005: * Permission to use, copy, modify, and distribute this software and its
006: * documentation for any purpose, without fee, and without written agreement is
007: * hereby granted, provided that the above copyright notice and the following
008: * two paragraphs appear in all copies of this software.
009: *
010: * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY FOR
011: * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT
012: * OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF THE UNIVERSITY OF
013: * CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
014: *
015: * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
016: * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
017: * AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
018: * ON AN "AS IS" BASIS, AND THE UNIVERSITY OF CALIFORNIA HAS NO OBLIGATION TO
019: * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
020: *
021: * Author: Matt Welsh <mdw@cs.berkeley.edu>
022: *
023: */
024:
025: package seda.sandStorm.lib.aSocket.nio;
026:
027: import seda.sandStorm.api.*;
028: import seda.sandStorm.core.*;
029: import seda.sandStorm.lib.aSocket.*;
030:
031: import java.nio.channels.*;
032: import java.nio.*;
033: import java.net.*;
034: import java.io.*;
035: import java.util.*;
036:
037: /**
038: * Internal class used to represent state of an active socket connection.
039: */
040: public class SockState extends seda.sandStorm.lib.aSocket.SockState {
041:
042: private static final boolean DEBUG = false;
043:
044: private SelectionKey rselkey, wselkey;
045: private ByteBuffer byte_buffer, read_byte_buffer;
046:
047: private NIOSelectSource read_selsource, write_selsource;
048:
049: SockState(ATcpConnection conn, Socket nbsock, int writeClogThreshold)
050: throws IOException {
051: if (DEBUG)
052: System.err.println("SockState: Constructor called with "
053: + conn + ", " + nbsock + ", " + writeClogThreshold);
054: this .conn = conn;
055: this .nbsock = nbsock;
056: this .writeClogThreshold = writeClogThreshold;
057: this .write_selsource = null;
058:
059: if (DEBUG)
060: System.err.println("SockState " + nbsock
061: + ": Const creating readBuf of size "
062: + aSocketConst.READ_BUFFER_SIZE);
063: readBuf = new byte[aSocketConst.READ_BUFFER_SIZE];
064: read_byte_buffer = ByteBuffer.wrap(readBuf);
065:
066: if (DEBUG)
067: System.err.println("SockState " + nbsock
068: + ": Setting flags");
069: outstanding_writes = 0;
070: numEmptyWrites = 0;
071: writeReqList = new ssLinkedList();
072:
073: clogged_qel = null;
074: clogged_numtries = 0;
075: if (DEBUG)
076: System.err.println("SockState " + nbsock + ": Const done");
077: }
078:
079: public SockState(ATcpConnection conn, Socket nbsock,
080: Integer writeClogThreshold) throws IOException {
081: this (conn, nbsock, writeClogThreshold.intValue());
082: }
083:
084: // This is synchronized with close()
085: protected synchronized void readInit(SelectSourceIF read_selsource,
086: SinkIF compQ, int readClogTries) {
087: if (DEBUG)
088: System.err.println("readInit called on " + this );
089: if (closed)
090: return; // May have been closed already
091: this .read_selsource = (NIOSelectSource) read_selsource;
092: this .read_selsource.setName("ReadSelectSource");
093: this .readCompQ = compQ;
094: this .readClogTries = readClogTries;
095: if (DEBUG)
096: System.err.println("n_keys = "
097: + ((NIOSelectSource) read_selsource).getSelector()
098: .keys().size());
099: rselkey = (SelectionKey) read_selsource.register(nbsock
100: .getChannel(), SelectionKey.OP_READ);
101: if (rselkey == null) {
102: System.err.println("SockState: register returned null");
103: return;
104: }
105: rselkey.attach(this );
106: }
107:
108: protected void doRead() {
109: if (DEBUG)
110: System.err.println("SockState: doRead called");
111:
112: // When using SelectSource, we need this guard, since after closing
113: // a socket we may have outstanding read events still in the queue
114: if (closed)
115: return;
116:
117: if (clogged_qel != null) {
118: // Try to drain the clogged element first
119: if (DEBUG)
120: System.err
121: .println("SockState: doRead draining clogged element "
122: + clogged_qel);
123: try {
124: readCompQ.enqueue(clogged_qel);
125: clogged_qel = null;
126: clogged_numtries = 0;
127: } catch (SinkFullException qfe) {
128: // Nope, still clogged
129: if ((readClogTries != -1)
130: && (++clogged_numtries >= readClogTries)) {
131: if (DEBUG)
132: System.err
133: .println("SockState: warning: readClogTries exceeded, dropping "
134: + clogged_qel);
135: clogged_qel = null;
136: clogged_numtries = 0;
137: } else {
138: // Try again later
139: return;
140: }
141: } catch (SinkException sce) {
142: // Whoops - user went away - just drop
143: this .close(null);
144: }
145: }
146:
147: int len;
148:
149: try {
150: if (DEBUG)
151: System.err.println("SockState: doRead trying read");
152: len = nbsock.getChannel().read(read_byte_buffer);
153: if (DEBUG)
154: System.err.println("SockState: read returned " + len);
155:
156: if (len == 0) {
157: // XXX MDW: Sometimes we get an error return result from
158: // poll() which causes an attempted read here, but no
159: // IOException. For now I am going to just drop the "null"
160: // packet - on Linux it seems that certain TCP errors can
161: // trigger this.
162: // System.err.println("ss.doRead: Warning: Got empty read on socket");
163: return;
164: } else if (len < 0) {
165: // Read failed - assume socket is dead
166: if (DEBUG)
167: System.err
168: .println("ss.doRead: read failed, sock closed");
169: this .close(readCompQ);
170: return;
171: }
172: } catch (Exception e) {
173: // Read failed - assume socket is dead
174: if (DEBUG)
175: System.err.println("ss.doRead: read got IOException: "
176: + e.getMessage());
177: this .close(readCompQ);
178: return;
179: }
180:
181: if (DEBUG)
182: System.err
183: .println("ss.doRead: Pushing up new ATcpInPacket, len="
184: + len);
185:
186: pkt = new ATcpInPacket(conn, readBuf, len,
187: aSocketConst.READ_BUFFER_COPY, seqNum);
188: // 0 is special (indicates no sequence number)
189: seqNum++;
190: if (seqNum == 0)
191: seqNum = 1;
192: if (aSocketConst.READ_BUFFER_COPY == false) {
193: readBuf = new byte[aSocketConst.READ_BUFFER_SIZE];
194: read_byte_buffer = ByteBuffer.wrap(readBuf);
195: }
196:
197: try {
198: readCompQ.enqueue(pkt);
199: } catch (SinkFullException qfe) {
200: clogged_qel = pkt;
201: clogged_numtries = 0;
202: return;
203: } catch (SinkException sce) {
204: // User has gone away
205: this .close(null);
206: return;
207: }
208: read_byte_buffer.rewind();
209: }
210:
211: // XXX This is synchronized with close() to avoid a race with close()
212: // removing the writeReqList while this method is being called.
213: // Probably a better way to do this...
214: protected synchronized boolean addWriteRequest(aSocketRequest req,
215: SelectSourceIF write_selsource) {
216: if (closed)
217: return false;
218:
219: if (DEBUG)
220: System.err.println("SockState: addWriteRequest called");
221:
222: if (this .write_selsource == null) {
223: if (DEBUG)
224: System.err.println("SockState: Setting selsource to "
225: + write_selsource);
226: if (DEBUG)
227: System.err.println("w/r="
228: + ((NIOSelectSource) write_selsource)
229: .getSelector()
230: + "/"
231: + ((NIOSelectSource) read_selsource)
232: .getSelector());
233: if (DEBUG)
234: System.err.println("n_keys = "
235: + ((NIOSelectSource) write_selsource)
236: .getSelector().keys().size());
237: this .write_selsource = (NIOSelectSource) write_selsource;
238: this .write_selsource.setName("WriteSelectSource");
239: wselkey = (SelectionKey) write_selsource.register(nbsock
240: .getChannel(), SelectionKey.OP_WRITE);
241: if (wselkey == null) {
242: System.err.println("SockState: register returned null");
243: return false;
244: }
245: wselkey.attach(this );
246: numActiveWriteSockets++;
247: if (DEBUG)
248: System.err
249: .println("SockState: Registered with selsource");
250: } else if (this .outstanding_writes == 0) {
251: numEmptyWrites = 0;
252: writeMaskEnable();
253: }
254:
255: if ((writeClogThreshold != -1)
256: && (this .outstanding_writes > writeClogThreshold)) {
257: if (DEBUG)
258: System.err
259: .println("SockState: warning: writeClogThreshold exceeded, dropping "
260: + req);
261: if (req instanceof ATcpWriteRequest)
262: return false;
263: if (req instanceof ATcpCloseRequest) {
264: // Do immediate close: Assume socket is clogged
265: ATcpCloseRequest creq = (ATcpCloseRequest) req;
266: this .close(creq.compQ);
267: return true;
268: }
269: }
270:
271: if (DEBUG)
272: System.err.println("SockState: Adding writeReq to tail");
273: writeReqList.add_to_tail(req);
274: this .outstanding_writes++;
275: if (DEBUG)
276: System.err.println("SockState: " + this .outstanding_writes
277: + " outstanding writes");
278: return true;
279: }
280:
281: protected void initWrite(ATcpWriteRequest req) {
282: this .cur_write_req = req;
283: this .writeBuf = req.buf.data;
284: this .cur_offset = req.buf.offset;
285: this .cur_length_target = req.buf.size + cur_offset;
286: this .byte_buffer = ByteBuffer.wrap(writeBuf, cur_offset,
287: req.buf.size);
288: }
289:
290: protected boolean tryWrite() throws SinkClosedException {
291: try {
292: int tryLen;
293: if (DEBUG)
294: System.err.println("SockState: tryWrite()");
295: if (MAX_WRITE_LEN == -1) {
296: tryLen = cur_length_target - cur_offset;
297: } else {
298: tryLen = Math.min(cur_length_target - cur_offset,
299: MAX_WRITE_LEN);
300: }
301: if (DEBUG)
302: System.err.println("writing " + tryLen + " bytes");
303: byte_buffer.limit(byte_buffer.position() + tryLen);
304: cur_offset += nbsock.getChannel().write(byte_buffer);
305: if (DEBUG)
306: System.err.println("SockState: tryWrite() of " + tryLen
307: + " bytes (len=" + cur_length_target + ", off="
308: + cur_offset);
309:
310: } catch (IOException ioe) {
311: // Assume this is because socket was already closed
312: this .close(null);
313: throw new SinkClosedException(
314: "tryWrite got exception doing write: "
315: + ioe.getMessage());
316: }
317: if (cur_offset == cur_length_target) {
318: if (DEBUG)
319: System.err
320: .println("SockState: tryWrite() completed write of "
321: + cur_length_target + " bytes");
322: return true;
323: } else
324: return false;
325: }
326:
327: protected void writeMaskEnable() {
328: numActiveWriteSockets++;
329: wselkey.interestOps(wselkey.interestOps()
330: | SelectionKey.OP_WRITE);
331: }
332:
333: protected void writeMaskDisable() {
334: numActiveWriteSockets--;
335: wselkey.interestOps(wselkey.interestOps()
336: & ~SelectionKey.OP_WRITE);
337: }
338:
339: // XXX This is synchronized to avoid close() interfering with
340: // addWriteRequest
341: protected synchronized void close(SinkIF closeEventQueue) {
342: if (closed)
343: return;
344:
345: closed = true;
346:
347: if (DEBUG)
348: System.err
349: .println("SockState.close(): Deregistering with selsources");
350: if (read_selsource != null)
351: read_selsource.deregister(rselkey);
352: if (write_selsource != null)
353: write_selsource.deregister(rselkey);
354: if (DEBUG)
355: System.err
356: .println("SockState.close(): done deregistering with selsources");
357: // Eliminate write queue
358:
359: // XXX XXX XXX MDW: This introduces a race condition with
360: // addWriteRequest() -- need to serialize close() with other
361: // queue operations on the socket.
362: writeReqList = null;
363:
364: try {
365: if (DEBUG)
366: System.err.println("SockState.close(): doing close ["
367: + nbsock + "]");
368: nbsock.close();
369: } catch (IOException e) {
370: // Do nothing
371: }
372:
373: if (closeEventQueue != null) {
374: SinkClosedEvent sce = new SinkClosedEvent(conn);
375: closeEventQueue.enqueue_lossy(sce);
376: }
377: }
378:
379: }
|