Track the actual item count in BlockingCollection

src/libraries/System.Collections.Concurrent/src/System/Collections/Concurrent/BlockingCollection.cs

@@ -45,6 +45,7 @@ namespace System.Collections.Concurrent
     public class BlockingCollection<T> : IEnumerable<T>, ICollection, IDisposable, IReadOnlyCollection<T>
     {
         private IProducerConsumerCollection<T> _collection;
+        private int _actualCount;
         private int _boundedCapacity;
         private const int NON_BOUNDED = -1;
         private SemaphoreSlim? _freeNodes;
@@ -92,7 +93,7 @@ public bool IsCompleted
             get
             {
                 CheckDisposed();
-                return (IsAddingCompleted && (_occupiedNodes.CurrentCount == 0));
+                return (IsAddingCompleted && (Volatile.Read(ref _actualCount) == 0));
             }
         }
 
@@ -105,7 +106,7 @@ public int Count
             get
             {
                 CheckDisposed();
-                return _occupiedNodes.CurrentCount;
+                return Volatile.Read(ref _actualCount);
             }
         }
 
@@ -212,6 +213,7 @@ private void Initialize(IProducerConsumerCollection<T> collection, int boundedCa
             Debug.Assert(boundedCapacity > 0 || boundedCapacity == NON_BOUNDED);
 
             _collection = collection;
+            _actualCount = collectionCount;
             _boundedCapacity = boundedCapacity;
             _isDisposed = false;
             _consumersCancellationTokenSource = new CancellationTokenSource();
@@ -480,9 +482,18 @@ private bool TryAddWithNoTimeValidation(T item, int millisecondsTimeout, Cancell
                 finally
                 {
                     if (addingSucceeded)
+                    {
+                        int current;
+                        do
+                        {
+                            current = Volatile.Read(ref _actualCount);
+                        }
+                        while (Interlocked.CompareExchange(ref _actualCount, current + 1, current) != current);
+
                         //After adding an element to the underlying storage, signal to the consumers
                         //waiting on _occupiedNodes that there is a new item added ready to be consumed.
                         _occupiedNodes.Release();
+                    }
                     else
                         //TryAdd did not result in increasing the size of the underlying store and hence we need
                         //to increment back the count of the _freeNodes semaphore.
@@ -706,6 +717,13 @@ private bool TryTakeWithNoTimeValidation([MaybeNullWhen(false)] out T item, int
                     // removeFaulted implies !removeSucceeded, but the reverse is not true.
                     if (removeSucceeded)
                     {
+                        int current;
+                        do
+                        {
+                            current = Volatile.Read(ref _actualCount);
+                        }
+                        while (Interlocked.CompareExchange(ref _actualCount, current - 1, current) != current);
+
                         if (_freeNodes != null)
                         {
                             Debug.Assert(_boundedCapacity != NON_BOUNDED);

src/libraries/System.Collections.Concurrent/tests/BlockingCollectionTests.cs

@@ -704,6 +704,38 @@ public static void Test13_IsSynchronized_SyncRoot()
                 "Test13_IsSynchronized_SyncRoot: > test failed - IsSynchronized should be false");
         }
 
+        /// <summary>
+        /// Validates that BlockingCollection.IsCompleted remains false after cancellation when items are still available.
+        /// </summary>
+        /// <returns>True if test succeeded, false otherwise.</returns>
+        [Fact]
+        public static void Test14_IsCompleted_RemainsFalse_AfterCancellation()
+        {
+            BlockingCollection<int> blockingCollection = ConstructBlockingCollection<int>();
+            CancellationTokenSource cts = new CancellationTokenSource();
+
+            blockingCollection.Add(10);
+            blockingCollection.CompleteAdding();
+
+            Assert.False(blockingCollection.IsCompleted);
+
+            Task consumer = Task.Run(() =>
+            {
+                Assert.Throws<OperationCanceledException>(() => blockingCollection.Take(cts.Token));
+            });
+
+            cts.Cancel();
+            consumer.Wait();
+
+            Assert.False(blockingCollection.IsCompleted);
+
+            int item;
+            Assert.True(blockingCollection.TryTake(out item));
+            Assert.Equal(10, item);
+
+            Assert.True(blockingCollection.IsCompleted);
+        }
+
         /// <summary>Initializes an array of blocking collections such that all are full except one in case of Adds and
         /// all are empty except one (the same blocking collection) in case of Takes.
         /// Adds "numOfAdds" elements to the BlockingCollection and then takes "numOfTakes" elements and checks