聊聊同步机制

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}


private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    // The code below is very delicate, to achieve these goals:
    // - call nanoTime exactly once for each call to park
    // - if nanos <= 0L, return promptly without allocation or nanoTime
    // - if nanos == Long.MIN_VALUE, don't underflow
    // - if nanos == Long.MAX_VALUE, and nanoTime is non-monotonic
    //   and we suffer a spurious wakeup, we will do no worse than
    //   to park-spin for a while
    long startTime = 0L;    // Special value 0L means not yet parked
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        else if (s == COMPLETING)
            // We may have already promised (via isDone) that we are done
            // so never return empty-handed or throw InterruptedException
            Thread.yield();
        else if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        else if (q == null) {
            if (timed && nanos <= 0L)
                return s;
            q = new WaitNode();
        }
        else if (!queued)
            queued = U.compareAndSwapObject(this, WAITERS,
                                            q.next = waiters, q);
        else if (timed) {
            final long parkNanos;
            if (startTime == 0L) { // first time
                startTime = System.nanoTime();
                if (startTime == 0L)
                    startTime = 1L;
                parkNanos = nanos;
            } else {
                long elapsed = System.nanoTime() - startTime;
                if (elapsed >= nanos) {
                    removeWaiter(q);
                    return state;
                }
                parkNanos = nanos - elapsed;
            }
            // nanoTime may be slow; recheck before parking
            if (state < COMPLETING)
                LockSupport.parkNanos(this, parkNanos);
        }
        else
            LockSupport.park(this);
    }
}
复制

public void run() {
    if (state != NEW ||
        !U.compareAndSwapObject(this, RUNNER, null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                set(result);
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}


protected void set(V v) {
    if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
        outcome = v;
        U.putOrderedInt(this, STATE, NORMAL); // final state
        finishCompletion();
    }
}


private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (U.compareAndSwapObject(this, WAITERS, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }
    done();
    callable = null;        // to reduce footprint
}
复制

public void park(boolean absolute, long time) {
      if (absolute) {
          Thread.currentThread().parkUntil$(time);
       } else {
           Thread.currentThread().parkFor$(time);
       }
   }
复制

public final void parkFor$(long nanos) {
       synchronized(lock) {
       switch (parkState) {
           case ParkState.PREEMPTIVELY_UNPARKED: {
               parkState = ParkState.UNPARKED;
               break;
           }
           case ParkState.UNPARKED: {
               long millis = nanos  NANOS_PER_MILLI;
               nanos %= NANOS_PER_MILLI;


               parkState = ParkState.PARKED;
               try {
                   lock.wait(millis, (int) nanos);
               } catch (InterruptedException ex) {
                   interrupt();
               } finally {
                   /*
                    * Note: If parkState manages to become
                    * PREEMPTIVELY_UNPARKED before hitting this
                    * code, it should left in that state.
                    */
                   if (parkState == ParkState.PARKED) {
                       parkState = ParkState.UNPARKED;
                   }
               }
               break;
           }
           default /*parked*/: {
               throw new AssertionError("Attempt to repark");
           }
       }
       }
   }
复制

private final CountDownLatch resolved;
public T getValue() throws InvocationTargetException, InterruptedException {
    this.resolved.await();
    if (this.fail == null) {
        return this.value;
    } else {
        throw new InvocationTargetException(this.fail);
    }
}


void resolve(T v, Throwable f) {
    synchronized(this.resolved) {
        if (this.resolved.getCount() == 0L) {
            throw new IllegalStateException("Already resolved");
        }


        this.value = v;
        this.fail = f;
        this.resolved.countDown();
    }


    this.notifyCallbacks();
}
复制

# coding: utf-8
# blocking  epoll io client
import selectorsimport socket
import time


sel = selectors.DefaultSelector()
def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready，拷贝到内存空间，并分配一个新的socket，给当前接入的请求
    print('accepted', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read) # 将这个sokcet注册到 sel，也就是sel也可以开始监听这个sock，并注册回调，等待下次轮训
def read(conn, mask):
    data = conn.recv(1000)  # 下次轮训，执行这个回调
    if data:
        print('echoing', conn)
        conn.send(data) # 回传数据给请求方。
        print('send over', conn)


    else:
        print('closing', conn) 
        sel.unregister(conn) # 请求结束，删除socket监听
        conn.close()


sock = socket.socket() # 绑定一个socket
sock.bind(('localhost', 12325))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept) # selector注册监听这个socket。可以把这个sokcet当作一个流。一直监听的只有这个接收请求的socket
i = 1
while True: # 轮训sel的ready事件
    events = sel.select() 
    time.sleep(10)
    for key, mask in events: 
        callback = key.data
        callback(key.fileobj, mask)
    print('this is ', i)
    i += 1
复制