a-foundation/src/main/java/com/ajjpj/afoundation/concurrent/SharedQueueNonBlockingImpl.java
@Override public void push (Runnable task) {
        while (true) {
            final long _base = UNSAFE.getLongVolatile (this, OFFS_BASE);
            final long _top = top;

            if (_top == _base + mask) {
                throw new RejectedExecutionExceptionWithoutStacktrace ("Queue overflow");
            }

            final long taskOffset = taskOffset (_top);
            if (UNSAFE.compareAndSwapObject (tasks, taskOffset, null, task)) {
                // if the publishing thread is interrupted here, other publishers will effectively do a spin wait
                if (!UNSAFE.compareAndSwapLong (this, OFFS_TOP, _top, _top+1)) {
                    // there was a buffer wrap-around in the meantime --> undo the CAS 'put' operation and try again
                    UNSAFE.putObjectVolatile (tasks, taskOffset, null);
                    continue;
                }

                if (_top - _base <= 1) {
                    pool.onAvailableTask ();
                }
                break;
            }
        }
    }
@Override public void push (Runnable task) {
        while (true) {
            final long _base = UNSAFE.getLongVolatile (this, OFFS_BASE);
            final long _top = top;

            if (_top == _base + mask) {
                throw new RejectedExecutionExceptionWithoutStacktrace ("Queue overflow");
            }

            final long taskOffset = taskOffset (_top);
            if (UNSAFE.compareAndSwapObject (tasks, taskOffset, null, task)) {
                // if the publishing thread is interrupted here, other publishers will effectively do a spin wait
                if (!UNSAFE.compareAndSwapLong (this, OFFS_TOP, _top, _top+1)) {
                    // there was a buffer wrap-around in the meantime --> undo the CAS 'put' operation and try again
                    UNSAFE.putObjectVolatile (tasks, taskOffset, null);
                    continue;
                }

                if (_top - _base <= 1) {
                    pool.onAvailableTask ();
                }
                break;
            }
        }
    }
@ Override public void push ( Runnable task ) { while ( true ) { final long _base = UNSAFE . getLongVolatile ( this , OFFS_BASE ) ; final long _top = top ; if ( _top == _base + mask ) { throw new RejectedExecutionExceptionWithoutStacktrace ( "Queue overflow" ) ; } final long taskOffset = taskOffset ( _top ) ; if ( UNSAFE . compareAndSwapObject ( tasks , taskOffset , null , task ) ) { // if the publishing thread is interrupted here, other publishers will effectively do a spin wait if ( ! UNSAFE . compareAndSwapLong ( this , OFFS_TOP , _top , _top + 1 ) ) { // there was a buffer wrap-around in the meantime --> undo the CAS 'put' operation and try again UNSAFE . putObjectVolatile ( tasks , taskOffset , null ) ; continue ; } if ( _top - _base <= 1 ) { pool . onAvailableTask ( ) ; } break ; } } }
Add a new task to the top of the shared queue, incrementing 'top'.
Add a new task to the top of the shared queue incrementing top .