fix(windows,server): 提交键慢响应改为分片让出等待,不再丢字或冻结宿主 (#982)

* fix(windows,server): 提交键慢响应改为分片让出等待,不再丢字或冻结宿主

* fix(windows): 提交回复轮询不阻塞宿主,补齐标点顶候选路径并保持结算顺序

- 标点顶高亮候选(WM_AsyncPunctuationCommit 无预取文本)也在处理函数里分片取回复,
  edit session 不再在宿主 UI 线程上阻塞 300 ms
- 轮询改为不阻塞:先查缓存、PeekNamedPipe,有帧才以零超时读,不再每片阻塞 16 ms
- 定时器只重派已开始等待的请求,按请求 id 升序结算,较新的提交排在仍在等的旧提交之后
- 处理函数取到 TransportUnavailable 时直接按歧义丢键并重置,不再交给 edit session
- 已在等回复的提交键在同一焦点下被组合纪元取代时按 DeliveryAmbiguous 处理,避免两端分叉
- 回复缓存淘汰不再淘汰刚写入的条目;CachePendingReply 与提交回复缓存共用一份逻辑
- 停用时显式 KillTimer(TIMER_COMMIT_REPLY_POLL);windows/AGENTS.md 同步新的提交预算

---------

Co-authored-by: fanlusky <1730976608@qq.com>
This commit is contained in:
Jia Suiandfanlusky authored and GitHub committed 2026-10-16 02:48:58 +08:00
1 parent 64fdc775f0
commit 9beab473f4
9 files changed
+478 -40

No files matched your search

+10 -10
View File
@@ -8,16 +8,16 @@ namespace FanyImeIpc
// asynchronously painted CandidatePageSnapshot. Pin-frequency reorders a page after every commit, so
// a selection that runs ahead of the painted frame commits a candidate the user never saw.
//
// The wait bound is load-bearing: the TSF side reads a committing reply with
// FANY_IME_COMMIT_REPLY_TIMEOUT_MS (300ms, windows/src/IPC/Ipc.h) and on a miss tears the pipe
// down and makes the user choose again. The wait is woken by the render echo, so a condition
// variable timeout (rounded up to one timer tick, ~16ms) is the only slack on top of this bound;
// together with candidate resolution it must stay well inside that budget. The frequency write
// runs after the reply and no longer competes for it. On a slow machine (battery, WebView2 paint)
// a paint regularly takes longer than 30ms, and settling against the unpainted page is exactly the
// "committed something I did not see" bug, so the bound leaves room for a slow paint. On timeout
// the caller continues with the current data and the diagnostic log records the miss. Do not
// raise this without re-deriving the TSF budget.
// The wait bound is load-bearing: the TSF side waits for a committing reply within
// FANY_IME_COMMIT_REPLY_TIMEOUT_MS (1500ms, windows/src/IPC/Ipc.h) and at the end of that window
// tears the pipe down and makes the user choose again. The wait is woken by the render echo, so a
// condition variable timeout (rounded up to one timer tick, ~16ms) is the only slack on top of
// this bound; together with candidate resolution it must stay well inside that budget. The
// frequency write runs after the reply and no longer competes for it. On a slow machine (battery,
// WebView2 paint) a paint regularly takes longer than 30ms, and settling against the unpainted
// page is exactly the "committed something I did not see" bug, so the bound leaves room for a slow
// paint. On timeout the caller continues with the current data and the diagnostic log records the
// miss. Do not raise this without re-deriving the TSF budget.
constexpr int kCandidateSelectionRenderWaitMaxMs = 80;
// Header-only pure policy so tests can pin it without linking the Windows/server stack.
@@ -43,7 +43,7 @@ TEST_CASE(candidate_render_sync_wait_is_bounded_to_a_visible_stall)
{
// The bound is a UX contract and a transport constraint: a wedged UI thread may delay a
// committed keystroke by at most this, and the wait must stay well inside the TSF commit reply
// timeout (300ms, windows/src/IPC/Ipc.h FANY_IME_COMMIT_REPLY_TIMEOUT_MS) or TSF tears the pipe
// down and the user has to choose again.
// timeout (1500ms, windows/src/IPC/Ipc.h FANY_IME_COMMIT_REPLY_TIMEOUT_MS) or TSF tears the
// pipe down and the user has to choose again.
REQUIRE_EQ(FanyImeIpc::kCandidateSelectionRenderWaitMaxMs, 80);
}
+8 -2
View File
@@ -95,11 +95,17 @@
|---|---|---|
| Stale | focus token、组合纪元、reset token 或焦点代已被取代 | 只丢这颗键,不再发起重置 |
| Resync | 宿主拒绝 edit session(S_FALSE、E_NOTIMPL 等)、RequestEditSession 失败、异步投递失败,以及 OnEndEdit 组合属性检查、文档不匹配等非传输问题 | 丢键、吞队列、本地取消组合;不轮换 token、不关管道,闸门重开时在同一 token 上发 HideCandidateWnd 清 Server 组合 |
| Transport | 写失败、回复读失败/坏帧、`TransportUnavailable`、`DefinitelyNotSent`、`DeliveryAmbiguous`、提交回复 300 ms 超时、worker 断开 | 丢键、吞队列、完整重置(新 focus token、重开回复管道) |
| Transport | 写失败、回复读失败/坏帧、`TransportUnavailable`、`DefinitelyNotSent`、`DeliveryAmbiguous`、提交回复 1500 ms 窗口耗尽、已在等回复的提交键在同一焦点下被组合纪元取代、worker 断开 | 丢键、吞队列、完整重置(新 focus token、重开回复管道) |
| Offline | 离线原始通道里的宿主失败 | 丢键、吞队列,保留本地组合(此时它是唯一权威) |
翻页与高亮移动键(PageUp/PageDown/Up/Down/Tab)的回复只是确认,缺失是软错过,不拆管道、不清
队列;其余候选键可能上屏,保留 300 ms 提交预算与「歧义即丢键 + 传输重置」。
队列;其余候选键可能上屏,保留「歧义即丢键 + 传输重置」。提交回复的总窗口是
`FANY_IME_COMMIT_REPLY_TIMEOUT_MS`(1500 ms),但等待跑在宿主 UI 线程上,不能一次阻塞读完:
消息处理函数用 `_PollCommitReply` 做不阻塞的轮询,没到就把请求放回表里、由
`TIMER_COMMIT_REPLY_POLL` 唤醒再试,并按请求 id 顺序结算。会进 edit session 的提交路径(确定、
数字选词、标点顶高亮候选)在处理函数里先取回复、按请求 id 缓存,session 直接消费;session 自己的
单次等待只剩 `FANY_IME_COMMIT_REPLY_SESSION_WAIT_MS`(300 ms)兜底。新增会上屏的异步路径要接入
同一套轮询,不要在 session 里阻塞等回复。
## 焦点、候选与离线回退
+1
View File
@@ -1083,6 +1083,7 @@ STDAPI CMetasequoiaIME::Deactivate()
KillTimer(_msgWndHandle, TIMER_DEFERRED_FOCUS_LOSS);
KillTimer(_msgWndHandle, TIMER_FOCUS_STATUS_RESEND);
KillTimer(_msgWndHandle, TIMER_FOCUS_CARET_STATE);
KillTimer(_msgWndHandle, TIMER_COMMIT_REPLY_POLL);
// 不同于上面几个:成对标点的重试定时器还带着 _pairedCaretRetryTimerActive
// 这一份状态,只 KillTimer 会让标志停在 true,消息窗口重建后就再也装不上
// 定时器了,所以走完整的取消路径。
+40
View File
@@ -141,6 +141,16 @@ constexpr UINT_PTR TIMER_DEFERRED_FOCUS_LOSS = 4;
constexpr UINT_PTR TIMER_FOCUS_STATUS_RESEND = 5;
constexpr UINT_PTR TIMER_PAIRED_PUNCTUATION_CARET = 6;
constexpr UINT_PTR TIMER_FOCUS_CARET_STATE = 7;
// Re-arms a committing key's reply poll between slices. The wake-up must be a
// timer, not a message posted to self: GetMessage only generates WM_PAINT and
// WM_TIMER while the thread queue is empty, so a self-posted re-dispatch keeps
// the queue non-empty for the whole window and defers every paint the host is
// waiting on (measured: zero WM_PAINT and zero WM_TIMER during a one-second
// re-post loop, all of them delivered the moment it drained). WM_TIMER ranks
// below WM_PAINT, so with a timer the host repaints between slices.
// The id must not collide with the constants above.
constexpr UINT_PTR TIMER_COMMIT_REPLY_POLL = 8;
constexpr UINT COMMIT_REPLY_POLL_INTERVAL_MS = 1;
// Runs after the focus status resend and the reconnect it may trigger, so the
// Server has activated this client before the badge request arrives.
constexpr UINT FOCUS_CARET_STATE_DELAY_MS = 120;
@@ -475,11 +485,41 @@ class CMetasequoiaIME : public ITfTextInputProcessorEx,
uint64_t compositionEpoch = 0;
uint64_t deferredReplayToken = 0;
std::wstring prefetchedText;
// End of the cooperative commit-reply window (GetTickCount64 domain),
// 0 until the first slice starts it. Kept on the request so a re-posted
// slice resumes the same window instead of restarting it.
ULONGLONG commitReplyDeadlineTick = 0;
};
bool _PostAsyncKeyRequest(UINT message, UINT code, WCHAR wch, uint64_t requestId, std::wstring prefetchedText = {},
uint64_t expectedFocusToken = 0, uint64_t expectedCompositionEpoch = 0,
uint64_t deferredReplayToken = 0);
bool _TakeAsyncKeyRequest(UINT message, UINT token, _Out_ AsyncKeyRequest &request);
// One slice of a committing key's reply wait. The wait runs on the host UI
// thread, so instead of blocking for the whole window this polls without
// blocking, re-queues the request on "not ready", and returns false so the
// handler yields the message pump. A request that is behind an older
// waiting commit yields without reading, so commits settle in send order.
// Returns true with *receivedData set to the reply or to a
// TransportUnavailable frame.
bool _PollCommitReply(UINT message, UINT token, _Inout_ AsyncKeyRequest &request,
_Outptr_result_maybenull_ FanyImeNamedpipeDataToTsf **receivedData);
bool _RequeueAsyncKeyRequest(UINT token, UINT message, AsyncKeyRequest request);
// Runs on TIMER_COMMIT_REPLY_POLL: re-dispatches, oldest request first, the
// requests that yielded waiting for their committing reply, and stops the
// timer once none is left.
void _DispatchPendingCommitReplyPolls();
// Fetches the reply a commit key's edit session would otherwise wait for and
// caches it by request id, so the session consumes it at once. Returns false
// when the caller must not invoke the session: it yielded, or the reply was
// a transport failure that has already been settled here.
bool _FetchCommitReplyForEditSession(UINT message, UINT token, AsyncKeyRequest &request);
// A delivered commit key whose reply is lost: the Server may already have
// committed, so it is dropped and never resent, with a transport reset.
void _FailUnansweredCommitKey(uint64_t requestId, uint64_t deferredReplayToken);
// The request's focus token or composition epoch was replaced. A commit
// key that already yielded waiting for its reply was delivered, so on the
// same focus it fails as unanswered instead of being dropped as stale.
void _FailSupersededAsyncKeyRequest(const AsyncKeyRequest &request);
struct WorkerCandidateCommit
{
std::wstring text;
@@ -30,6 +30,17 @@ using namespace metasequoia_ime_detail;
namespace
{
// Messages whose handler waits for a committing reply and re-queues its request
// when the reply is not in yet (see _PollCommitReply). Only a request whose
// wait has started (commitReplyDeadlineTick != 0) is re-dispatched by the
// timer; one still in the posted queue runs from its own message.
bool IsCommitReplyPollMessage(UINT message)
{
return message == WM_AsyncServerCandidateKey || message == WM_AsyncCycleInputMode ||
message == WM_AsyncFinalizeCandidate || message == WM_AsyncNumberCandidateCommit ||
message == WM_AsyncPunctuationCommit;
}
class CPunctuationCommitEditSession : public CEditSessionBase
{
public:
@@ -462,6 +473,200 @@ bool CMetasequoiaIME::_TakeAsyncKeyRequest(UINT message, UINT token, _Out_ Async
return true;
}
bool CMetasequoiaIME::_RequeueAsyncKeyRequest(UINT token, UINT message, AsyncKeyRequest request)
{
const HWND ownerWindow = _msgWndHandle;
if (token == 0 || !ownerWindow || !IsWindow(ownerWindow))
{
return false;
}
{
std::lock_guard<std::mutex> lock(_pendingCommitCandidateMutex);
_pendingAsyncKeyMessages[token] = std::move(request);
}
// Wake on a timer instead of re-posting the message: a posted message keeps
// the thread queue non-empty for the whole wait, which defers every WM_PAINT
// and WM_TIMER the host is waiting on. A timer yields the same slice and
// still lets the host repaint (see TIMER_COMMIT_REPLY_POLL).
if (SetTimer(ownerWindow, TIMER_COMMIT_REPLY_POLL, COMMIT_REPLY_POLL_INTERVAL_MS, nullptr) != 0)
{
return true;
}
// No timer slot left: the posted message is the pre-timer behaviour and
// only costs the host its repaints, so it stays as the fallback.
if (PostMessage(ownerWindow, message, static_cast<WPARAM>(token), 0))
{
return true;
}
std::lock_guard<std::mutex> lock(_pendingCommitCandidateMutex);
_pendingAsyncKeyMessages.erase(token);
return false;
}
void CMetasequoiaIME::_DispatchPendingCommitReplyPolls()
{
const HWND ownerWindow = _msgWndHandle;
if (!ownerWindow || !IsWindow(ownerWindow))
{
return;
}
struct Poll
{
uint64_t requestId;
UINT token;
UINT message;
};
std::vector<Poll> polls;
{
std::lock_guard<std::mutex> lock(_pendingCommitCandidateMutex);
for (const auto &entry : _pendingAsyncKeyMessages)
{
// A request still in the posted queue has not started waiting and
// runs from its own message; dispatching it here would run it ahead
// of the messages posted before it.
if (IsCommitReplyPollMessage(entry.second.message) && entry.second.commitReplyDeadlineTick != 0)
{
polls.push_back({entry.second.requestId, entry.first, entry.second.message});
}
}
}
// Commits settle in the order they were sent, not in token order.
std::sort(polls.begin(), polls.end(), [](const Poll &a, const Poll &b) { return a.requestId < b.requestId; });
// Synchronous dispatch: the timer pass already yielded to the host, and
// returning with an empty queue is what lets the system generate the next
// WM_PAINT. Every handler either settles its request or re-arms the timer.
// Nothing is held while dispatching: the handlers take the same mutex.
bool stillPolling = false;
for (const Poll &poll : polls)
{
SendMessage(ownerWindow, poll.message, static_cast<WPARAM>(poll.token), 0);
std::lock_guard<std::mutex> lock(_pendingCommitCandidateMutex);
const auto entry = _pendingAsyncKeyMessages.find(poll.token);
if (entry != _pendingAsyncKeyMessages.end() && entry->second.commitReplyDeadlineTick != 0)
{
// Still waiting: the newer commits stay behind it.
stillPolling = true;
break;
}
}
if (!stillPolling)
{
// Self-limiting: with nothing left to poll the timer would otherwise
// wake this thread once a tick for the life of the message window.
KillTimer(ownerWindow, TIMER_COMMIT_REPLY_POLL);
}
}
bool CMetasequoiaIME::_PollCommitReply(UINT message, UINT token, _Inout_ AsyncKeyRequest &request,
_Outptr_result_maybenull_ FanyImeNamedpipeDataToTsf **receivedData)
{
*receivedData = nullptr;
if (request.commitReplyDeadlineTick == 0)
{
request.commitReplyDeadlineTick = GetTickCount64() + FANY_IME_COMMIT_REPLY_TIMEOUT_MS;
}
// An older commit that is still waiting settles first. The Server answers in
// send order, so its reply is ahead of this one and this one cannot be
// applied before it; the older key's deadline is also the earlier one.
bool behindOlderCommit = false;
{
std::lock_guard<std::mutex> lock(_pendingCommitCandidateMutex);
for (const auto &entry : _pendingAsyncKeyMessages)
{
if (IsCommitReplyPollMessage(entry.second.message) && entry.second.commitReplyDeadlineTick != 0 &&
entry.second.requestId < request.requestId)
{
behindOlderCommit = true;
break;
}
}
}
// Past its own deadline the slice reads anyway, so this key can never wait
// longer than its window.
if (!behindOlderCommit || GetTickCount64() >= request.commitReplyDeadlineTick)
{
FanyImeNamedpipeDataToTsf *reply = nullptr;
const CommitReplyWaitStatus status =
TryReadCommitReplySlice(request.requestId, request.commitReplyDeadlineTick, &reply);
if (status != CommitReplyWaitStatus::NotReady)
{
*receivedData = reply;
return true;
}
}
// Yield instead of blocking: keep the request pending and wake on the poll
// timer, so the host message pump runs until the next slice. The fresh
// dispatch revalidates the token, the focus token and the composition
// epoch before touching state.
const uint64_t requestId = request.requestId;
const uint64_t deferredReplayToken = request.deferredReplayToken;
if (_RequeueAsyncKeyRequest(token, message, std::move(request)))
{
return false;
}
// The request was delivered but can no longer be re-posted, so its reply may
// never be observed. That is the same "delivered but unanswered" state the
// transport branch reports: drop the key and never resend it.
_FailUnansweredCommitKey(requestId, deferredReplayToken);
return false;
}
bool CMetasequoiaIME::_FetchCommitReplyForEditSession(UINT message, UINT token, AsyncKeyRequest &request)
{
FanyImeNamedpipeDataToTsf *receivedData = nullptr;
if (!_PollCommitReply(message, token, request, &receivedData))
{
return false;
}
if (receivedData->msg_type == Global::DataFromServerMsgType::TransportUnavailable)
{
// Settled here rather than handed to the session: the transport is
// already being reset, which may clear the reply cache before the
// session runs, and a session that misses the cache would block the
// host again and tear down the freshly reconnected pipe.
_FailUnansweredCommitKey(request.requestId, request.deferredReplayToken);
return false;
}
// The session that applies the key must not wait for the same reply again.
CacheCommitReplyForEditSession(request.requestId, *receivedData);
return true;
}
void CMetasequoiaIME::_FailUnansweredCommitKey(uint64_t requestId, uint64_t deferredReplayToken)
{
const DeferredKeyFailureReason reason = IsDeliveredServerRequestId(requestId)
? DeferredKeyFailureReason::DeliveryAmbiguous
: DeferredKeyFailureReason::TransportBroken;
if (deferredReplayToken != 0)
{
_FailDeferredKey(deferredReplayToken, reason);
}
else
{
_ResetSessionAfterFailure(DeferredKeyFailureKind::Transport);
}
}
void CMetasequoiaIME::_FailSupersededAsyncKeyRequest(const AsyncKeyRequest &request)
{
// A key that yielded waiting for its commit reply was delivered: the Server
// may already have committed while the host ended the composition. Dropping
// it as stale would leave the Server ahead of TSF. A focus change already
// discarded everything the key belonged to, so that one stays stale.
if (request.commitReplyDeadlineTick != 0 && _IsFocusSessionCurrent(request.focusToken))
{
_FailUnansweredCommitKey(request.requestId, request.deferredReplayToken);
return;
}
_FailDeferredKey(request.deferredReplayToken, DeferredKeyFailureReason::Superseded);
}
void CMetasequoiaIME::_ClearAsyncKeyRequests()
{
std::vector<uint64_t> deferredReplayTokens;
+58 -9
View File
@@ -400,6 +400,15 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
break;
}
case WM_TIMER: {
if (wParam == TIMER_COMMIT_REPLY_POLL)
{
// A committing key could not settle its reply in the last slice, so
// the poll re-armed here. WM_TIMER is lower priority than WM_PAINT,
// which is exactly the point: the host repaints while this key
// waits, instead of after it.
pIME->_DispatchPendingCommitReplyPolls();
break;
}
if (wParam == TIMER_DEFERRED_FOCUS_LOSS)
{
KillTimer(hWnd, TIMER_DEFERRED_FOCUS_LOSS);
@@ -649,7 +658,14 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
if (!pIME->_IsFocusSessionCurrent(request.focusToken) ||
!pIME->_IsCompositionEpochCurrent(request.compositionEpoch))
{
pIME->_FailDeferredKey(request.deferredReplayToken, DeferredKeyFailureReason::Superseded);
pIME->_FailSupersededAsyncKeyRequest(request);
break;
}
// Fetch the committing reply here, where the message pump can be
// yielded, and leave it for the edit session below. Otherwise the
// session would block the host UI thread waiting for the same reply.
if (!pIME->_FetchCommitReplyForEditSession(WM_AsyncFinalizeCandidate, static_cast<UINT>(wParam), request))
{
break;
}
PerfTimer timer;
@@ -686,7 +702,20 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
if (!pIME->_IsFocusSessionCurrent(request.focusToken) ||
!pIME->_IsCompositionEpochCurrent(request.compositionEpoch))
{
pIME->_FailDeferredKey(request.deferredReplayToken, DeferredKeyFailureReason::Superseded);
pIME->_FailSupersededAsyncKeyRequest(request);
break;
}
// With no prefetched text, punctuation that commits the highlighted
// candidate reads that candidate's reply in the edit session
// (_HandleCompositionPunctuation). Fetch it here, where the message
// pump can be yielded, under the same condition the session uses; a
// wait that already started keeps going even if that state changed.
const bool commitsHighlightedCandidate =
request.prefetchedText.empty() && pIME->_candidateMode != CANDIDATE_NONE &&
pIME->_pCandidateListUIPresenter && Global::CommitWithHighlightedCandPunc.count(request.wch) > 0;
if ((request.commitReplyDeadlineTick != 0 || commitsHighlightedCandidate) &&
!pIME->_FetchCommitReplyForEditSession(WM_AsyncPunctuationCommit, static_cast<UINT>(wParam), request))
{
break;
}
PerfTimer timer;
@@ -738,10 +767,16 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
if (!pIME->_IsFocusSessionCurrent(request.focusToken) ||
!pIME->_IsCompositionEpochCurrent(request.compositionEpoch))
{
pIME->_FailDeferredKey(request.deferredReplayToken, DeferredKeyFailureReason::Superseded);
pIME->_FailSupersededAsyncKeyRequest(request);
break;
}
FanyImeNamedpipeDataToTsf *receivedData = nullptr;
if (!pIME->_PollCommitReply(WM_AsyncCycleInputMode, static_cast<UINT>(wParam), request, &receivedData))
{
// The request was re-queued and re-posted (or already failed): the
// handler yields the message pump without settling.
break;
}
FanyImeNamedpipeDataToTsf *receivedData = TryReadCommitReplyFromServerPipe(request.requestId);
// The edit handler parses the payload once and rejects malformed text
// through the same deferred-cycle failure policy.
if (receivedData->msg_type != Global::DataFromServerMsgType::TrilingualCycle)
@@ -793,7 +828,7 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
if (!pIME->_IsFocusSessionCurrent(request.focusToken) ||
!pIME->_IsCompositionEpochCurrent(request.compositionEpoch))
{
pIME->_FailDeferredKey(request.deferredReplayToken, DeferredKeyFailureReason::Superseded);
pIME->_FailSupersededAsyncKeyRequest(request);
break;
}
const UINT code = request.code;
@@ -804,9 +839,17 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
// Every other candidate key may commit the highlighted candidate, so
// it gets the commit reply budget and is never resent once delivered.
const bool navigationOnly = IsCandidateNavigationOnlyKey(code);
FanyImeNamedpipeDataToTsf *receivedData =
navigationOnly ? TryReadDataFromServerPipeWithTimeout(request.requestId, /*abortTransportOnTimeout=*/false)
: TryReadCommitReplyFromServerPipe(request.requestId);
FanyImeNamedpipeDataToTsf *receivedData = nullptr;
if (navigationOnly)
{
receivedData = TryReadDataFromServerPipeWithTimeout(request.requestId, /*abortTransportOnTimeout=*/false);
}
else if (!pIME->_PollCommitReply(WM_AsyncServerCandidateKey, static_cast<UINT>(wParam), request, &receivedData))
{
// The request was re-queued and re-posted (or already failed): the
// handler yields the message pump without settling.
break;
}
if (receivedData->msg_type == Global::DataFromServerMsgType::TransportUnavailable)
{
// A transport failure is not text and must never be committed to
@@ -890,7 +933,13 @@ LRESULT CALLBACK CMetasequoiaIME_WindowProc(HWND hWnd, UINT message, WPARAM wPar
if (!pIME->_IsFocusSessionCurrent(request.focusToken) ||
!pIME->_IsCompositionEpochCurrent(request.compositionEpoch))
{
pIME->_FailDeferredKey(request.deferredReplayToken, DeferredKeyFailureReason::Superseded);
pIME->_FailSupersededAsyncKeyRequest(request);
break;
}
// Fetch the committing reply here, where the message pump can be
// yielded; the edit session below then consumes it without blocking.
if (!pIME->_FetchCommitReplyForEditSession(WM_AsyncNumberCandidateCommit, static_cast<UINT>(wParam), request))
{
break;
}
PerfTimer timer;
+110 -9
View File
@@ -314,6 +314,22 @@ bool IsValidServerReply(const FanyImeNamedpipeDataToTsf &reply)
return false;
}
// Keeps the cache bounded by evicting the oldest request ids, but never the
// entry just stored: that is the reply a waiter is about to look up.
void StorePendingReply(uint64_t requestId, const FanyImeNamedpipeDataToTsf &reply)
{
pendingReplies[requestId] = reply;
auto oldest = pendingReplies.begin();
while (pendingReplies.size() > MaxPendingReplyCount)
{
if (oldest->first == requestId)
{
++oldest;
}
oldest = pendingReplies.erase(oldest);
}
}
void CachePendingReply(const FanyImeNamedpipeDataToTsf &reply)
{
if (reply.msg_type == Global::DataFromServerMsgType::PipeReady ||
@@ -330,11 +346,7 @@ void CachePendingReply(const FanyImeNamedpipeDataToTsf &reply)
}
return;
}
pendingReplies[reply.request_id] = reply;
while (pendingReplies.size() > MaxPendingReplyCount)
{
pendingReplies.erase(pendingReplies.begin());
}
StorePendingReply(reply.request_id, reply);
}
bool WriteOverlappedWithTimeout(HANDLE hPipeHandle, const void *data, DWORD size, DWORD &bytesWritten)
@@ -1329,22 +1341,96 @@ bool SendToNamedpipe(bool *deliveryAmbiguous = nullptr)
static struct FanyImeNamedpipeDataToTsf *TryReadDataFromServerPipeWithDeadline(uint64_t expectedRequestId,
bool abortTransportOnTimeout,
DWORD timeoutMs);
DWORD timeoutMs, bool *notReady);
struct FanyImeNamedpipeDataToTsf *TryReadDataFromServerPipeWithTimeout(uint64_t expectedRequestId,
bool abortTransportOnTimeout)
{
return TryReadDataFromServerPipeWithDeadline(expectedRequestId, abortTransportOnTimeout, 50);
return TryReadDataFromServerPipeWithDeadline(expectedRequestId, abortTransportOnTimeout, 50, nullptr);
}
struct FanyImeNamedpipeDataToTsf *TryReadCommitReplyFromServerPipe(uint64_t expectedRequestId)
{
return TryReadDataFromServerPipeWithDeadline(expectedRequestId, true, FANY_IME_COMMIT_REPLY_TIMEOUT_MS);
return TryReadDataFromServerPipeWithDeadline(expectedRequestId, true, FANY_IME_COMMIT_REPLY_SESSION_WAIT_MS,
nullptr);
}
// True when a slice has something to consume right now: a cached reply, a
// frame waiting in the pipe, or a state only the read path can report (no
// request id, no pipe, a dead pipe). False means the read would only block.
static bool HasCommitReplyToRead(uint64_t expectedRequestId)
{
if (expectedRequestId == FANY_IME_NO_REQUEST_ID || pendingReplies.count(expectedRequestId) != 0 ||
!IsValidPipeHandle(hFromServerPipe))
{
return true;
}
// The reply pipe has no read outstanding between slices, so this is the
// same nonblocking probe TryOpenClientPipe uses.
DWORD bytesAvailable = 0;
return !PeekNamedPipe(hFromServerPipe, nullptr, 0, nullptr, &bytesAvailable, nullptr) || bytesAvailable != 0;
}
CommitReplyWaitStatus TryReadCommitReplySlice(uint64_t expectedRequestId, ULONGLONG deadlineTick,
struct FanyImeNamedpipeDataToTsf **reply)
{
constexpr ULONGLONG kWindowMs = FANY_IME_COMMIT_REPLY_TIMEOUT_MS;
*reply = nullptr;
// A slice never blocks the host UI thread: it reads only what is already
// in the pipe (zero read timeout) and otherwise yields at once. The timer
// between slices is what spaces the polls out.
bool notReady = true;
struct FanyImeNamedpipeDataToTsf *fetched = nullptr;
if (HasCommitReplyToRead(expectedRequestId))
{
notReady = false;
fetched =
TryReadDataFromServerPipeWithDeadline(expectedRequestId, /*abortTransportOnTimeout=*/true, 0, &notReady);
}
if (!notReady)
{
*reply = fetched;
return fetched->msg_type == Global::DataFromServerMsgType::TransportUnavailable
? CommitReplyWaitStatus::Transport
: CommitReplyWaitStatus::Ready;
}
if (GetTickCount64() < deadlineTick)
{
return CommitReplyWaitStatus::NotReady;
}
// The total window is over with no reply. The request was delivered, so the
// Server may already have committed the selection: take exactly the
// transport reset a single blocking miss always took.
const ULONGLONG startedAt = deadlineTick >= kWindowMs ? deadlineTick - kWindowMs : 0;
DebugTsfKeyLatency(L"reply-id-timeout", expectedRequestId, static_cast<double>(GetTickCount64() - startedAt),
HRESULT_FROM_WIN32(ERROR_TIMEOUT));
ClosePipeHandleIfValid(hFromServerPipe);
RequestNamedpipeReconnect();
transportUnavailableReply = {};
transportUnavailableReply.msg_type = Global::DataFromServerMsgType::TransportUnavailable;
*reply = &transportUnavailableReply;
return CommitReplyWaitStatus::Transport;
}
void CacheCommitReplyForEditSession(uint64_t expectedRequestId, const struct FanyImeNamedpipeDataToTsf &reply)
{
if (expectedRequestId == FANY_IME_UNSOLICITED_REQUEST_ID || expectedRequestId == FANY_IME_NO_REQUEST_ID ||
reply.msg_type == Global::DataFromServerMsgType::PipeReady ||
FanyImeProtocol::IsNegotiationReply(reply.msg_type))
{
return;
}
// Keyed by the request the session asked for, and never the entry evicted
// to keep the cache bounded.
StorePendingReply(expectedRequestId, reply);
}
static struct FanyImeNamedpipeDataToTsf *TryReadDataFromServerPipeWithDeadline(uint64_t expectedRequestId,
bool abortTransportOnTimeout,
DWORD timeoutMs)
DWORD timeoutMs, bool *notReady)
{
PerfTimer replyWaitTimer;
@@ -1402,6 +1488,14 @@ static struct FanyImeNamedpipeDataToTsf *TryReadDataFromServerPipeWithDeadline(u
hFromServerPipe, &namedpipeDataFromServer, sizeof(namedpipeDataFromServer), remainingMs, bytesRead);
if (readResult == OverlappedReadResult::TimedOut)
{
if (notReady)
{
// A cooperative commit poll: this slice ended with no reply,
// but the pipe is healthy and the total window is not over.
// Leave the pipe, the reply cache and the composition alone.
*notReady = true;
return nullptr;
}
DebugTsfKeyLatency(L"reply-wait-timeout", expectedRequestId, replyWaitTimer.ElapsedMs(),
HRESULT_FROM_WIN32(ERROR_TIMEOUT));
if (!abortTransportOnTimeout)
@@ -1464,6 +1558,13 @@ static struct FanyImeNamedpipeDataToTsf *TryReadDataFromServerPipeWithDeadline(u
// Server may already have committed a selection, so preserving local
// composition would split the two state machines. Treat this exactly like
// a broken transport and schedule a tokenized reset plus local cancel.
if (notReady)
{
// Same as the read-timeout above: a cooperative slice boundary, not a
// failure. The reply cache was updated, the pipe stays up.
*notReady = true;
return nullptr;
}
if (!abortTransportOnTimeout)
{
DebugTsfKeyLatency(L"reply-soft-miss", expectedRequestId, replyWaitTimer.ElapsedMs(), S_FALSE);
+44 -8
View File
@@ -119,15 +119,51 @@ bool TryReadCandidatePageFromSharedMemory(_Out_ std::wstring *candidatePage);
// returns a non-TransportUnavailable empty frame for the caller to fall back.
struct FanyImeNamedpipeDataToTsf *TryReadDataFromServerPipeWithTimeout(uint64_t expectedRequestId,
bool abortTransportOnTimeout);
// Replies that commit text (candidate selection, candidate + punctuation) get
// a longer budget than ordinary keys. The request has already been written, so
// a miss is DeliveryAmbiguous: the Server may have committed the selection.
// Waiting for the same request_id is the only way to learn which candidate it
// chose; a slow machine (battery, throttled disk) routinely needs more than
// the ordinary 50ms. A miss past this budget drops the key and resets the
// transport; the selection is never sent a second time.
constexpr DWORD FANY_IME_COMMIT_REPLY_TIMEOUT_MS = 300;
// Replies that commit text (candidate selection, candidate + punctuation) are
// authoritative: the request has already been written, so a miss is
// DeliveryAmbiguous and the Server may have committed the selection. Waiting
// for the same request_id is the only way to learn which candidate it chose; a
// slow key-handling thread (measured at 984ms) routinely needs far more than
// the ordinary 50ms. That wait runs on the host UI thread, so this is the *total*
// window a committing key may spend waiting, not a single blocking read:
// callers poll it with nonblocking TryReadCommitReplySlice calls and yield the
// message pump between them. Only a real transfer failure, or the end of this
// window, tears the transport down; the selection is never sent a second time.
constexpr DWORD FANY_IME_COMMIT_REPLY_TIMEOUT_MS = 1500;
// An edit session that applies a commit key does not wait that window: the
// message handler already fetched the reply and cached it by request id (see
// CacheCommitReplyForEditSession), so the session consumes it at once. Every
// asynchronous commit path (finalize, number select, candidate + punctuation)
// fetches in its handler. This shorter bound is the pre-existing single-shot
// wait, kept for the synchronous fallback and for the race where a transport
// reset clears the cache in between. It must stay small enough that a miss
// cannot visibly freeze the host.
constexpr DWORD FANY_IME_COMMIT_REPLY_SESSION_WAIT_MS = 300;
struct FanyImeNamedpipeDataToTsf *TryReadCommitReplyFromServerPipe(uint64_t expectedRequestId);
// How one slice of a committing key's reply wait ended.
enum class CommitReplyWaitStatus
{
// *reply holds the authoritative reply (or a cached one).
Ready,
// No reply yet and the total window is not over. The pipe is left up, the
// reply is not consumed and nothing is torn down: the caller must yield
// (re-post its message) and poll again.
NotReady,
// A real transfer failure, or the total window expired. *reply holds a
// TransportUnavailable frame; the caller takes the existing reset path.
Transport,
};
// Polls the commit reply for expectedRequestId once without blocking: it only
// consumes frames already in the pipe, and reports the window over once
// deadlineTick (GetTickCount64 domain) has passed. *reply is set for Ready and
// Transport.
CommitReplyWaitStatus TryReadCommitReplySlice(uint64_t expectedRequestId, ULONGLONG deadlineTick,
_Out_ struct FanyImeNamedpipeDataToTsf **reply);
// Puts a reply a message handler already fetched back into the pending-reply
// cache, keyed by the request the edit session will look up. The session that
// applies the key then consumes it without waiting again. Transport failures
// are not cached: the handler settles those itself.
void CacheCommitReplyForEditSession(uint64_t expectedRequestId, const struct FanyImeNamedpipeDataToTsf &reply);
// True when requestId names a request that was actually written to the Server,
// i.e. a lost reply is DeliveryAmbiguous rather than DefinitelyNotSent.
inline bool IsDeliveredServerRequestId(uint64_t requestId)