zchuango commented on code in PR #3509: URL: https://github.com/apache/brpc/pull/3509#discussion_r4246534634
########## src/brpc/ubshm/ub_cleanup_worker.cpp: ########## @@ -0,0 +1,153 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include <pthread.h> +#include <stdint.h> +#include "butil/atomicops.h" +#include "butil/containers/mpsc_queue.h" +#include "butil/logging.h" +#include "butil/scoped_lock.h" +#include "bthread/bthread.h" +#include "brpc/ubshm/ub_cleanup_worker.h" +#include "brpc/ubshm/ub_ring_manager.h" + +namespace brpc { +namespace ubring { + +namespace { + +// Cleanup latency is tolerated in seconds (it only delays releasing memory of +// an already dead link), so the idle worker polls instead of spinning and the +// module needs no extra wakeup primitive. +constexpr uint64_t kCleanupIdlePollIntervalUs = 100 * 1000; + +butil::atomic<bool> g_worker_started(false); +// Set when the worker bthread could not be created. Terminal: posting then +// fails forever, and every caller falls back to running its work inline -- the +// behaviour from before the worker existed. Never cleared, so no job can ever +// be queued behind a worker that will not run. +butil::atomic<bool> g_worker_unavailable(false); +// Posted-but-not-finished jobs. A poster increments it before enqueueing and +// the worker decrements it after the job returned, so DrainAndWait can wait +// for quiescence without observing the queue itself. +butil::atomic<int> g_cleanup_inflight(0); +butil::MPSCQueue<UbrCleanupJob> g_cleanup_queue; Review Comment: The cleanup queue now stays alive through process exit, so its destructor can’t race the worker’s Dequeue(). ########## src/brpc/ubshm/shm/shm_ubs.cpp: ########## @@ -434,30 +440,57 @@ void *UbsShmCallback(void* args, uint64_t) return nullptr; } - // Drain one node per fire and keep the SDK calls outside the lock, so - // a slow daemon cannot stall the timer thread for the whole list. + // The timer thread runs every timer in the process, so it must not call + // into the SDK: post the drain to the cleanup worker. At most one step may + // be in flight, otherwise a slow SDK call would pile up jobs for the same + // list head. + if (g_shm_drain_pending.exchange(true)) { + return nullptr; + } + if (BAIDU_UNLIKELY(!UbrCleanupWorker::PostShmDrain(shm_list, UbsShmDrainNode))) { + // The worker could not be started: keep the pre-worker behaviour and + // drain here rather than leaving the nodes in the list forever. + LOG(ERROR) << "Cleanup worker unavailable, draining shm on the timer thread."; + UbsShmDrainNode(shm_list); Review Comment: Worker startup is now retried on each post. A failed start falls back to inline cleanup for that post only; later posts can use the worker once startup succeeds. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
