[idd] clipboard: batch helper stream delivery

Extend the clipboard target boundary with ordered prefix delivery.
Pass up to four queued LGMP stream records to the Helper ring in one
operation, apply state only to the accepted prefix, and retain the
remaining suffix for exact retries.

The channel holds its write lock across the prefix and emits one pipe
doorbell per stream window instead of one per record.

idd/LGIdd/transport/CClipboardHub.cpp

@@ -384,14 +384,31 @@ void CClipboardHub::ClipboardReset(uint64_t epoch, uint32_t reason)
 ClipboardChannelResult CClipboardHub::SendClipboard(
   const KVMFRClipboardMessage& record, const uint8_t * data)
 {
-  // Keep the shared lock through Send so Unbind cannot advance the binding
-  // generation after validation but before the record enters the channel.
-  CSRWSharedLock lock(m_lock);
-  if (!ValidClientDirection(record) ||
-      m_stopped || !m_available || !m_active || !m_running || m_failed ||
-      !m_source || record.generation != m_generation)
+  const ClipboardChannelWrite write = { record, data };
+  size_t accepted = 0;
+  return SendClipboardBatch(&write, 1, accepted);
+}
+
+ClipboardChannelResult CClipboardHub::SendClipboardBatch(
+  const ClipboardChannelWrite * records, size_t count, size_t& accepted)
+{
+  accepted = 0;
+  if (!records || !count || count > KVMFR_CLIPBOARD_SLOT_COUNT)
     return ClipboardChannelResult::FAILED;
-  return m_channel.Send(record, data);
+
+  // Keep the shared lock through Send so Unbind cannot advance the binding
+  // generation after validation but before the records enter the channel.
+  CSRWSharedLock lock(m_lock);
+  if (m_stopped || !m_available || !m_active || !m_running || m_failed ||
+      !m_source)
+    return ClipboardChannelResult::FAILED;
+
+  for (size_t i = 0; i < count; ++i)
+    if (!ValidClientDirection(records[i].record) ||
+        records[i].record.generation != m_generation)
+      return ClipboardChannelResult::FAILED;
+
+  return m_channel.SendBatch(records, count, accepted);
 }
 
 void CClipboardHub::ClipboardReceiveReady()

idd/LGIdd/transport/CClipboardHub.h

@@ -73,6 +73,9 @@ public:
   ClipboardChannelResult SendClipboard(
     const KVMFRClipboardMessage& record,
     const uint8_t * data) override;
+  ClipboardChannelResult SendClipboardBatch(
+    const ClipboardChannelWrite * records,
+    size_t count, size_t& accepted) override;
   void ClipboardReceiveReady() override;
   void ClipboardFailed() override;
 };

idd/LGIdd/transport/IClipboardSource.h

@@ -42,6 +42,12 @@ public:
   virtual ClipboardChannelResult SendClipboard(
     const KVMFRClipboardMessage& record, const uint8_t * data) = 0;
 
+  // Sends an ordered prefix of client-to-Helper records. accepted reports
+  // how many records were consumed for every result; the source retains and
+  // retries an unaccepted suffix when the target returns BUSY.
+  virtual ClipboardChannelResult SendClipboardBatch(
+    const ClipboardChannelWrite * records, size_t count, size_t& accepted) = 0;
+
   // The source calls ClipboardReceiveReady after a Helper-to-client record
   // returned BUSY and it can accept an exact retry of that record.
   virtual void ClipboardReceiveReady() = 0;

idd/LGIdd/transport/lgmp/CLGMPClipboardTransport.cpp

@@ -596,7 +596,7 @@ void CLGMPClipboardTransport::QueueFileDisconnect()
 bool CLGMPClipboardTransport::QueueInternalTarget(
   const KVMFRClipboardMessage& record)
 {
-  if (!m_pendingTarget.valid && !m_streamTargetBusy &&
+  if (!m_pendingTarget.valid && !m_streamTargetPrepared &&
       !m_internalTargetCount)
     return BeginTarget(record, false);
   if (m_internalTargetCount == INTERNAL_TARGET_COUNT)
@@ -610,7 +610,7 @@ bool CLGMPClipboardTransport::QueueInternalTarget(
 
 bool CLGMPClipboardTransport::PumpInternalTarget()
 {
-  if (m_pendingTarget.valid || m_streamTargetBusy ||
+  if (m_pendingTarget.valid || m_streamTargetPrepared ||
       !m_internalTargetCount)
     return true;
 
@@ -625,26 +625,27 @@ void CLGMPClipboardTransport::ClearStreamTargets()
 {
   for (StreamTarget& target : m_streamTarget)
     target.Clear();
-  m_streamTargetHead  = 0;
-  m_streamTargetCount = 0;
-  m_streamTargetBusy  = false;
+  m_streamTargetHead     = 0;
+  m_streamTargetCount    = 0;
+  m_streamTargetPrepared = 0;
 }
 
 void CLGMPClipboardTransport::ClearQueuedStreamTargets()
 {
-  if (!m_streamTargetBusy)
+  if (!m_streamTargetPrepared)
   {
     ClearStreamTargets();
     return;
   }
 
-  // The BUSY head has already crossed the target boundary and must receive
-  // the same drain treatment as a BUSY PendingTarget. Only discard records
-  // which have not yet been presented to the target.
-  for (unsigned i = 1; i < m_streamTargetCount; ++i)
+  // The prepared prefix has already crossed the target boundary and must
+  // receive the same drain treatment as a BUSY PendingTarget. Only discard
+  // records which have not yet been presented to the target.
+  for (unsigned i = m_streamTargetPrepared;
+      i < m_streamTargetCount; ++i)
     m_streamTarget[
       (m_streamTargetHead + i) % STREAM_TARGET_COUNT].Clear();
-  m_streamTargetCount = 1;
+  m_streamTargetCount = m_streamTargetPrepared;
 }
 
 bool CLGMPClipboardTransport::QueueStreamTarget(
@@ -663,39 +664,92 @@ bool CLGMPClipboardTransport::QueueStreamTarget(
   return true;
 }
 
-void CLGMPClipboardTransport::PopStreamTarget()
+void CLGMPClipboardTransport::PopStreamTargets(unsigned count)
 {
-  StreamTarget& target = m_streamTarget[m_streamTargetHead];
-  target.Clear();
+  for (unsigned i = 0; i < count; ++i)
+    m_streamTarget[
+      (m_streamTargetHead + i) % STREAM_TARGET_COUNT].Clear();
   m_streamTargetHead =
-    (m_streamTargetHead + 1) % STREAM_TARGET_COUNT;
-  --m_streamTargetCount;
-  m_streamTargetBusy = false;
+    (m_streamTargetHead + count) % STREAM_TARGET_COUNT;
+  m_streamTargetCount -= count;
+  m_streamTargetPrepared = count < m_streamTargetPrepared ?
+    m_streamTargetPrepared - count : 0;
 }
 
-bool CLGMPClipboardTransport::PublishStreamTarget()
+CLGMPClipboardTransport::StreamDisposition
+CLGMPClipboardTransport::PreviewStreamTarget(
+  const KVMFRClipboardMessage& record, Transfer& transfer,
+  CLGMPClipboardFiles& files)
 {
-  StreamTarget& target = m_streamTarget[m_streamTargetHead];
+  if (record.type == KVMFR_CLIPBOARD_MESSAGE_DATA)
+  {
+    if (record.transfer == m_discardClientToHelper)
+      return StreamDisposition::DISCARD;
+    if (record.token ||
+        !kvmfrClipboardTransferFromHelper(record.transfer) ||
+        !ValidateChunk(transfer, record))
+      return StreamDisposition::INVALID;
+    AdvanceChunk(transfer, record);
+    return StreamDisposition::READY;
+  }
+
+  const CLGMPClipboardFiles::Direction direction =
+    CLGMPClipboardFiles::Direction::CLIENT_TO_HELPER;
+  if (files.IsStaleTerminal(record, direction))
+    return StreamDisposition::STALE;
+  if (!files.Validate(record, direction, 0, false))
+    return StreamDisposition::INVALID;
+  files.Apply(record, direction);
+  return StreamDisposition::READY;
+}
+
+bool CLGMPClipboardTransport::PublishStreamTargets()
+{
+  ClipboardChannelWrite writes[STREAM_TARGET_COUNT] = {};
+  for (unsigned i = 0; i < m_streamTargetPrepared; ++i)
+  {
+    StreamTarget& target = m_streamTarget[
+      (m_streamTargetHead + i) % STREAM_TARGET_COUNT];
+    writes[i].record = target.record;
+    writes[i].data   = target.record.length ? target.data : nullptr;
+  }
+
+  size_t accepted = 0;
   const ClipboardChannelResult result = m_target ?
-    m_target->SendClipboard(target.record,
-      target.record.length ? target.data : nullptr) :
+    m_target->SendClipboardBatch(writes, m_streamTargetPrepared,
+      accepted) :
     ClipboardChannelResult::FAILED;
+  if (accepted > m_streamTargetPrepared ||
+      (result == ClipboardChannelResult::ACCEPTED &&
+        accepted != m_streamTargetPrepared))
+  {
+    DEBUG_ERROR("Helper accepted an invalid clipboard stream prefix: "
+      "accepted=%zu offered=%u", accepted, m_streamTargetPrepared);
+    ReleaseOwner("invalid Helper clipboard batch result", false);
+    m_failed = true;
+    return true;
+  }
+
+  for (size_t i = 0; i < accepted; ++i)
+  {
+    const StreamTarget& target = m_streamTarget[
+      (m_streamTargetHead + i) % STREAM_TARGET_COUNT];
+    ApplyInbound(target.record);
+  }
+  if (accepted)
+  {
+    PopStreamTargets(static_cast<unsigned>(accepted));
+    if (m_ownerClientID && !m_ownerReleasing)
+      RenewLease();
+  }
+
   if (result == ClipboardChannelResult::BUSY)
   {
-    m_streamTargetBusy = true;
     if (m_ownerClientID)
       RenewLease();
     return true;
   }
 
-  if (result == ClipboardChannelResult::ACCEPTED)
-  {
-    ApplyInbound(target.record);
-    if (m_ownerClientID && !m_ownerReleasing)
-      RenewLease();
-  }
-  PopStreamTarget();
-
   if (result == ClipboardChannelResult::FAILED)
   {
     ReleaseOwner("Helper delivery failed", false);
@@ -704,86 +758,87 @@ bool CLGMPClipboardTransport::PublishStreamTarget()
   return true;
 }
 
-bool CLGMPClipboardTransport::RetryStreamTarget()
+bool CLGMPClipboardTransport::RetryStreamTargets()
 {
-  if (!m_streamTargetBusy || m_pendingTarget.valid)
+  if (!m_streamTargetPrepared || m_pendingTarget.valid)
     return true;
 
-  // The retained head is already validated and stamped. Retry it exactly;
+  // The retained prefix is already validated and stamped. Retry it exactly;
   // owner-release state may have changed while the target was BUSY.
-  m_streamTargetBusy = false;
-  return PublishStreamTarget();
-}
-
-bool CLGMPClipboardTransport::ProcessStreamTarget()
-{
-  StreamTarget& target = m_streamTarget[m_streamTargetHead];
-  const KVMFRClipboardMessage record = target.record;
-  if (!m_ownerClientID)
-  {
-    PopStreamTarget();
-    return true;
-  }
-  if (m_ownerReleasing && !m_releaseClearHelper)
-  {
-    PopStreamTarget();
-    return true;
-  }
-
-  if (record.type == KVMFR_CLIPBOARD_MESSAGE_DATA &&
-      record.transfer == m_discardClientToHelper)
-  {
-    PopStreamTarget();
-    return true;
-  }
-
-  const bool fileData =
-    record.type == KVMFR_CLIPBOARD_MESSAGE_FILE_DATA;
-  const bool staleFileData = fileData && m_files.IsStaleTerminal(
-    record, CLGMPClipboardFiles::Direction::CLIENT_TO_HELPER);
-  const bool validData = fileData ?
-    (staleFileData || m_files.Validate(record,
-      CLGMPClipboardFiles::Direction::CLIENT_TO_HELPER, 0, false)) :
-    (!record.token && kvmfrClipboardTransferFromHelper(record.transfer) &&
-      ValidateChunk(m_clientToHelper, record));
-  if (!validData)
-  {
-    PopStreamTarget();
-    ReleaseOwner("invalid clipboard stream data", true);
-    return true;
-  }
-
-  if (!m_ownerReleasing)
-    RenewLease();
-  if (staleFileData)
-  {
-    PopStreamTarget();
-    return true;
-  }
-
-  target.record.generation = m_endpointGeneration;
-  return PublishStreamTarget();
+  return PublishStreamTargets();
 }
 
 bool CLGMPClipboardTransport::PumpStreamTarget()
 {
-  while (!m_pendingTarget.valid && !m_streamTargetBusy &&
+  while (!m_pendingTarget.valid && !m_streamTargetPrepared &&
       m_streamTargetCount)
-    if (!ProcessStreamTarget())
-      return false;
+  {
+    if (!m_ownerClientID ||
+        (m_ownerReleasing && !m_releaseClearHelper))
+    {
+      PopStreamTargets(1);
+      continue;
+    }
+
+    // Dependent chunks are validated against the state the preceding chunk
+    // would establish, without committing that state before Helper accepts
+    // the corresponding prefix.
+    Transfer transfer = m_clientToHelper;
+    CLGMPClipboardFiles files = m_files;
+    StreamDisposition disposition = StreamDisposition::READY;
+    for (unsigned i = 0; i < m_streamTargetCount; ++i)
+    {
+      StreamTarget& target = m_streamTarget[
+        (m_streamTargetHead + i) % STREAM_TARGET_COUNT];
+      disposition = PreviewStreamTarget(target.record, transfer, files);
+      if (disposition != StreamDisposition::READY)
+        break;
+      target.record.generation = m_endpointGeneration;
+      ++m_streamTargetPrepared;
+    }
+
+    if (m_streamTargetPrepared)
+    {
+      if (!m_ownerReleasing)
+        RenewLease();
+      if (!PublishStreamTargets())
+        return false;
+      if (m_streamTargetPrepared || m_failed)
+        return true;
+      continue;
+    }
+
+    PopStreamTargets(1);
+    if (disposition == StreamDisposition::STALE)
+    {
+      if (!m_ownerReleasing)
+        RenewLease();
+      continue;
+    }
+    if (disposition == StreamDisposition::INVALID)
+      ReleaseOwner("invalid clipboard stream data", true);
+  }
   return true;
 }
 
 bool CLGMPClipboardTransport::DrainStream(bool& received)
 {
   received = false;
-  if (!m_ownerClientID || !m_clientToHostStream)
+  if (m_failed || !m_ownerClientID || !m_clientToHostStream)
     return true;
 
-  for (unsigned drained = 0;
-      drained < STREAM_DRAIN_MAX &&
-      m_streamTargetCount < STREAM_TARGET_COUNT; ++drained)
+  for (unsigned drained = 0; drained < STREAM_DRAIN_MAX;)
   {
+    if (m_streamTargetCount == STREAM_TARGET_COUNT)
+    {
+      // Submit only after the bounded staging window is full. A retained
+      // prefix or pending control keeps the window backpressured here.
+      if (!PumpStreamTarget())
+        return false;
+      if (m_streamTargetCount == STREAM_TARGET_COUNT)
+        return true;
+    }
+
     LGMPStreamBuffer buffer = {};
     LGMP_STATUS status =
       lgmpHostStreamReadPeek(m_clientToHostStream, &buffer);
@@ -794,11 +849,15 @@ bool CLGMPClipboardTransport::DrainStream(bool& received)
     {
       if (m_ownerReleasing && status == LGMP_ERR_STREAM_UNBOUND)
         break;
+      if (!PumpStreamTarget())
+        return false;
       ReleaseOwner("clipboard stream binding lost", true);
       return true;
     }
     if (status != LGMP_OK)
     {
+      if (!PumpStreamTarget())
+        return false;
       Fail("lgmpHostStreamReadPeek", status);
       return false;
     }
@@ -828,11 +887,12 @@ bool CLGMPClipboardTransport::DrainStream(bool& received)
     }
     if (!queued)
     {
+      if (!PumpStreamTarget())
+        return false;
       ReleaseOwner("invalid clipboard stream record", true);
       return true;
     }
-    if (!PumpStreamTarget())
-      return false;
+    ++drained;
   }
 
   return PumpStreamTarget();
@@ -1970,10 +2030,9 @@ void CLGMPClipboardTransport::Thread()
 
       // A restarted client can reuse dataset and transfer IDs. Deliver every
       // old-owner cleanup record before admitting messages from a new owner.
-      if (!RetryTarget() || !RetryStreamTarget() ||
+      if (!RetryTarget() || !RetryStreamTargets() ||
           !PumpInternalTarget() ||
-          !PumpStreamTarget() ||
-          (!m_streamTargetBusy && m_internalTargetCount == 0 &&
+          (!m_streamTargetCount && m_internalTargetCount == 0 &&
             !DrainMessage()) ||
           !DrainStream(streamReceived) ||
           !RetryOwnerRelease() ||

idd/LGIdd/transport/lgmp/CLGMPClipboardTransport.h

@@ -64,6 +64,14 @@ private:
     FAILED,
   };
 
+  enum class StreamDisposition
+  {
+    READY,
+    DISCARD,
+    STALE,
+    INVALID,
+  };
+
   struct Transfer
   {
     uint64_t transfer            = 0;
@@ -147,9 +155,9 @@ private:
   Transfer m_helperToClient;
   PendingTarget m_pendingTarget;
   StreamTarget m_streamTarget[STREAM_TARGET_COUNT];
-  unsigned m_streamTargetHead  = 0;
-  unsigned m_streamTargetCount = 0;
-  bool m_streamTargetBusy      = false;
+  unsigned m_streamTargetHead     = 0;
+  unsigned m_streamTargetCount    = 0;
+  unsigned m_streamTargetPrepared = 0;
   KVMFRClipboardMessage m_internalTarget[INTERNAL_TARGET_COUNT] = {};
   unsigned m_internalTargetCount = 0;
   CLGMPClipboardFiles m_files;
@@ -190,12 +198,14 @@ private:
   void ClearQueuedStreamTargets();
   bool QueueStreamTarget(const KVMFRClipboardMessage& record,
     const uint8_t * data);
-  void PopStreamTarget();
-  bool PublishStreamTarget();
-  bool RetryStreamTarget();
+  void PopStreamTargets(unsigned count);
+  StreamDisposition PreviewStreamTarget(
+    const KVMFRClipboardMessage& record, Transfer& transfer,
+    CLGMPClipboardFiles& files);
+  bool PublishStreamTargets();
+  bool RetryStreamTargets();
   bool PumpStreamTarget();
   bool DrainStream(bool& received);
-  bool ProcessStreamTarget();
 
   PLGMPMemory FindAvailable(PLGMPMemory (&memory)[MEMORY_COUNT]) const;
   PostResult PostForOwner(uint64_t udata, PLGMPMemory memory);