half working

benchmarks/build.gradle.kts

@@ -44,12 +44,12 @@ jmh {
     jmhVersion = catalogVersion("jmh")
 
 //    includes.set(listOf("""jayo\.benchmarks\.BufferLatin1Benchmark.*"""))
-//    includes.set(listOf("""jayo\.benchmarks\.BufferUtf8Benchmark.*"""))
+    includes.set(listOf("""jayo\.benchmarks\.BufferUtf8Benchmark.*"""))
 //    includes.set(listOf("""jayo\.benchmarks\.JsonSerializationBenchmark.*"""))
 //    includes.set(listOf("""jayo\.benchmarks\.SlowReaderBenchmark.*"""))
 //    includes.set(listOf("""jayo\.benchmarks\.SlowWriterBenchmark.*"""))
 //    includes.set(listOf("""jayo\.benchmarks\.SocketReaderBenchmark.*"""))
-    includes.set(listOf("""jayo\.benchmarks\.TcpAndJsonSerializationBenchmark.*"""))
+//    includes.set(listOf("""jayo\.benchmarks\.TcpAndJsonSerializationBenchmark.*"""))
 }
 
 tasks {

benchmarks/src/jmh/kotlin/jayo/benchmarks/BufferUtf8Benchmark.kt

@@ -50,10 +50,10 @@ open class BufferUtf8Benchmark {
         )
     }
 
-    @Param("20", "2000", "200000")
+    @Param("20"/*, "2000", "200000"*/)
     private var length = 0
 
-    @Param("ascii", "latin1", "utf8", "2bytes"/*, "3bytes", "4bytes", "bad"*/)
+    @Param("ascii"/*, "latin1", "utf8", "2bytes", "3bytes", "4bytes", "bad"*/)
     private lateinit var encoding: String
 
     private lateinit var jayoBuffer: Buffer
@@ -86,39 +86,39 @@ open class BufferUtf8Benchmark {
         okioDecode = tempOkio.snapshot()
     }
 
-    @Benchmark
-    fun writeUtf8Okio() {
-        okioBuffer.writeUtf8(encode)
-        okioBuffer.clear()
-    }
-
-    @Benchmark
-    fun readUtf8Okio(): Int {
-        okioBuffer.write(okioDecode)
-        return okioBuffer.readUtf8().length
-    }
-
-    @Benchmark
-    fun writeUtf8Jayo() {
-        jayoBuffer.write(encode)
-        jayoBuffer.clear()
-    }
-
-    @Benchmark
-    fun writeByteStringJayo() {
-        jayoBuffer.write(jayoDecode)
-        jayoBuffer.clear()
-    }
+//    @Benchmark
+//    fun writeUtf8Okio() {
+//        okioBuffer.writeUtf8(encode)
+//        okioBuffer.clear()
+//    }
+//
+//    @Benchmark
+//    fun readUtf8Okio(): Int {
+//        okioBuffer.write(okioDecode)
+//        return okioBuffer.readUtf8().length
+//    }
+//
+//    @Benchmark
+//    fun writeUtf8Jayo() {
+//        jayoBuffer.write(encode)
+//        jayoBuffer.clear()
+//    }
+//
+//    @Benchmark
+//    fun writeByteStringJayo() {
+//        jayoBuffer.write(jayoDecode)
+//        jayoBuffer.clear()
+//    }
 
     @Benchmark
     fun readUtf8StringJayo(): Int {
         jayoBuffer.write(jayoDecode)
         return jayoBuffer.readString().length
     }
 
-    @Benchmark
-    fun readUtf8Jayo(): Int {
-        jayoBuffer.write(jayoDecode)
-        return jayoBuffer.readUtf8().length()
-    }
+//    @Benchmark
+//    fun readUtf8Jayo(): Int {
+//        jayoBuffer.write(jayoDecode)
+//        return jayoBuffer.readUtf8().length()
+//    }
 }

core/src/main/java/jayo/internal/DeflaterRawWriter.java

@@ -54,7 +54,7 @@ public void write(final @NonNull Buffer reader, final long byteCount) {
         }
 
         var remaining = byteCount;
-        var head = _reader.segmentQueue.head;
+        var head = _reader.segmentQueue.head();
         assert head != null;
         while (remaining > 0L) {
             var headLimit = head.limit;

core/src/main/java/jayo/internal/GzipRawReader.java

@@ -210,7 +210,7 @@ private void updateCrc(final @NonNull SegmentQueue segmentQueue,
         var _offset = offset;
         var _byteCount = byteCount;
         // Skip segments that we aren't checksumming.
-        var segment = segmentQueue.head;
+        var segment = segmentQueue.head();
         assert segment != null;
         while (_offset >= segment.limit - segment.pos) {
             _offset -= (segment.limit - segment.pos);

core/src/main/java/jayo/internal/HashingUtils.java

@@ -36,7 +36,7 @@ private HashingUtils() {
         }
         try (rawReader; final var segmentQueue = new ReaderSegmentQueue(rawReader)) {
             var remaining = segmentQueue.expectSize(Long.MAX_VALUE);
-            var head = segmentQueue.head;
+            var head = segmentQueue.head();
             while (remaining > 0L) {
                 assert head != null;
                 final var currentLimit = head.limit;
@@ -75,7 +75,7 @@ private HashingUtils() {
         }
         try (rawReader; final var segmentQueue = new ReaderSegmentQueue(rawReader)) {
             var remaining = segmentQueue.expectSize(Long.MAX_VALUE);
-            var head = segmentQueue.head;
+            var head = segmentQueue.head();
             while (remaining > 0L) {
                 assert head != null;
                 final var currentLimit = head.limit;

core/src/main/java/jayo/internal/OutputStreamRawWriter.java

@@ -67,7 +67,7 @@ public void write(final @NonNull Buffer reader, final long byteCount) {
         }
 
         var remaining = byteCount;
-        var head = _reader.segmentQueue.head;
+        var head = _reader.segmentQueue.head();
         assert head != null;
         var headLimit = head.limit;
         while (remaining > 0L) {

core/src/main/java/jayo/internal/PeekRawReader.java

@@ -52,7 +52,7 @@ public PeekRawReader(final @NonNull Reader upstream) {
         this.upstream = Objects.requireNonNull(upstream);
         buffer = (RealBuffer) getBufferFromReader(upstream);
         buffer.segmentQueue.expectSize(1L);
-        final var bufferHead = buffer.segmentQueue.head;
+        final var bufferHead = buffer.segmentQueue.head();
         if (bufferHead != null) {
             this.expectedSegment = bufferHead;
             this.expectedPos = bufferHead.pos;
@@ -72,7 +72,7 @@ public long readAtMostTo(final @NonNull Buffer writer, final long byteCount) {
             throw new IllegalStateException("this peek reader is closed");
         }
 
-        final var bufferHead = buffer.segmentQueue.head;
+        final var bufferHead = buffer.segmentQueue.head();
         // Reader becomes invalid if there is an expected Segment and it and the expected position does not match the
         // current head and head position of the upstream buffer
         if (expectedSegment != null &&

core/src/main/java/jayo/internal/ReaderSegmentQueue.java

@@ -12,8 +12,6 @@
 import org.jspecify.annotations.Nullable;
 
 import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
 import static java.lang.System.Logger.Level.TRACE;
 
@@ -97,7 +95,6 @@ final static class Async extends ReaderSegmentQueue {
 
         // non-volatile because always used inside the lock
         private long expectedSize = 0;
-        private final Lock lock = new ReentrantLock();
         private final Condition expectingSize = lock.newCondition();
 
         private volatile @Nullable RuntimeException exception = null;

core/src/main/java/jayo/internal/RealAsyncTimeout.java

@@ -253,7 +253,7 @@ private void writeCancellable(RealBuffer reader, long byteCount, RealCancelToken
                 while (remaining > 0L) {
                     // Count how many bytes to write. This loop guarantees we split on a segment boundary.
                     var _toWrite = 0L;
-                    var segment = reader.segmentQueue.head;
+                    var segment = reader.segmentQueue.head();
                     while (_toWrite < TIMEOUT_WRITE_SIZE) {
                         assert segment != null;
                         final var segmentSize = segment.limit - segment.pos;

core/src/main/java/jayo/internal/RealBasicLock.java

@@ -6,6 +6,7 @@
 package jayo.internal;
 
 import jayo.tools.BasicLock;
+import org.jspecify.annotations.NonNull;
 import org.jspecify.annotations.Nullable;
 
 import java.lang.invoke.MethodHandles;
@@ -15,14 +16,18 @@
 public final class RealBasicLock implements BasicLock {
     @SuppressWarnings("FieldMayBeFinal")
     private volatile @Nullable Thread lockingThread = null;
+    @SuppressWarnings("FieldMayBeFinal")
+    private volatile @Nullable Thread conditionThread = null;
 
     // VarHandle mechanics
     private static final VarHandle LOCKING_THREAD;
+    private static final VarHandle CONDITION_THREAD;
 
     static {
         try {
             final var l = MethodHandles.lookup();
             LOCKING_THREAD = l.findVarHandle(RealBasicLock.class, "lockingThread", Thread.class);
+            CONDITION_THREAD = l.findVarHandle(RealBasicLock.class, "conditionThread", Thread.class);
         } catch (ReflectiveOperationException e) {
             throw new ExceptionInInitializerError(e);
         }
@@ -40,8 +45,31 @@ public void lock() {
     public void unlock() {
         final var currentThread = Thread.currentThread();
         final var lockingThread = (Thread) LOCKING_THREAD.compareAndExchangeRelease(this, currentThread, null);
-        if (lockingThread != currentThread) {
+        if (currentThread != lockingThread && lockingThread != null) {
             LockSupport.unpark(lockingThread);
         }
     }
+
+    @Override
+    public @NonNull BasicCondition newCondition() {
+        return new BasicCondition() {
+            @Override
+            public void await() {
+                if (!CONDITION_THREAD
+                        .weakCompareAndSetRelease(RealBasicLock.this, null, Thread.currentThread())) {
+                    throw new IllegalStateException("condition is already held");
+                }
+                unlock();
+                LockSupport.park();
+            }
+
+            @Override
+            public void signal() {
+                final var conditionThread = (Thread) CONDITION_THREAD.getAndSetRelease(RealBasicLock.this, null);
+                if (conditionThread != null) {
+                    LockSupport.unpark(conditionThread);
+                }
+            }
+        };
+    }
 }