diff --git a/.travis.yml b/.travis.yml
index a37cd44d5..99d119366 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -327,8 +327,8 @@ script:
 
   # ray operator tests
   - cd ./deploy/ray-operator/
-  - go build
-  - go test ./...
+  - ../../ci/suppress_output go build
+  - ../../ci/suppress_output go test ./...
   - cd ../..
 
   # random python tests TODO(ekl): these should be moved to bazel
diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx
index 9daa00060..92a11650e 100644
--- a/python/ray/_raylet.pyx
+++ b/python/ray/_raylet.pyx
@@ -562,6 +562,16 @@ cdef CRayStatus check_signals() nogil:
     return CRayStatus.OK()
 
 
+cdef void gc_collect() nogil:
+    with gil:
+        start = time.perf_counter()
+        num_freed = gc.collect()
+        end = time.perf_counter()
+        logger.info(
+            "gc.collect() freed {} refs in {} seconds".format(
+                num_freed, end - start))
+
+
 cdef shared_ptr[CBuffer] string_to_buffer(c_string& c_str):
     cdef shared_ptr[CBuffer] empty_metadata
     if c_str.size() == 0:
@@ -607,7 +617,7 @@ cdef class CoreWorker:
             raylet_socket.encode("ascii"), job_id.native(),
             gcs_options.native()[0], log_dir.encode("utf-8"),
             node_ip_address.encode("utf-8"), node_manager_port,
-            task_execution_handler, check_signals, True))
+            task_execution_handler, check_signals, gc_collect, True))
 
     def run_task_loop(self):
         with nogil:
@@ -760,6 +770,10 @@ cdef class CoreWorker:
             check_status(self.core_worker.get().Delete(
                 free_ids, local_only, delete_creating_tasks))
 
+    def global_gc(self):
+        with nogil:
+            self.core_worker.get().TriggerGlobalGC()
+
     def set_object_store_client_options(self, client_name,
                                         int64_t limit_bytes):
         try:
diff --git a/python/ray/includes/libcoreworker.pxd b/python/ray/includes/libcoreworker.pxd
index 579008e9b..d51c812a3 100644
--- a/python/ray/includes/libcoreworker.pxd
+++ b/python/ray/includes/libcoreworker.pxd
@@ -97,6 +97,7 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
                         c_vector[shared_ptr[CRayObject]] *returns,
                         const CWorkerID &worker_id) nogil,
                     CRayStatus() nogil,
+                    void() nogil,
                     c_bool ref_counting_enabled)
         CWorkerType &GetWorkerType()
         CLanguage &GetLanguage()
@@ -173,6 +174,7 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
                         int64_t timeout_ms, c_vector[c_bool] *results)
         CRayStatus Delete(const c_vector[CObjectID] &object_ids,
                           c_bool local_only, c_bool delete_creating_tasks)
+        CRayStatus TriggerGlobalGC()
         c_string MemoryUsageString()
 
         CWorkerContext &GetWorkerContext()
diff --git a/python/ray/internal/internal_api.py b/python/ray/internal/internal_api.py
index a66cba1fb..d685cf7b3 100644
--- a/python/ray/internal/internal_api.py
+++ b/python/ray/internal/internal_api.py
@@ -1,7 +1,14 @@
 import ray.worker
 from ray import profiling
 
-__all__ = ["free"]
+__all__ = ["free", "global_gc"]
+
+
+def global_gc():
+    """Trigger gc.collect() on all workers in the cluster."""
+
+    worker = ray.worker.get_global_worker()
+    worker.core_worker.global_gc()
 
 
 def free(object_ids, local_only=False, delete_creating_tasks=False):
diff --git a/python/ray/tests/test_reference_counting.py b/python/ray/tests/test_reference_counting.py
index c6ab3527d..9eea21634 100644
--- a/python/ray/tests/test_reference_counting.py
+++ b/python/ray/tests/test_reference_counting.py
@@ -3,9 +3,11 @@ import copy
 import json
 import logging
 import os
+import gc
 import tempfile
 import time
 import uuid
+import weakref
 
 import numpy as np
 
@@ -14,6 +16,7 @@ import pytest
 import ray
 import ray.cluster_utils
 import ray.test_utils
+from ray.internal.internal_api import global_gc
 
 logger = logging.getLogger(__name__)
 
@@ -72,6 +75,46 @@ def check_refcounts(expected, timeout=10):
                 time.sleep(0.1)
 
 
+def test_global_gc(shutdown_only):
+    cluster = ray.cluster_utils.Cluster()
+    for _ in range(2):
+        cluster.add_node(num_cpus=1, num_gpus=0)
+    ray.init(address=cluster.address)
+
+    class ObjectWithCyclicRef:
+        def __init__(self):
+            self.loop = self
+
+    @ray.remote(num_cpus=1)
+    class GarbageHolder:
+        def __init__(self):
+            gc.disable()
+            x = ObjectWithCyclicRef()
+            self.garbage = weakref.ref(x)
+
+        def has_garbage(self):
+            return self.garbage() is not None
+
+    try:
+        gc.disable()
+
+        # Local driver.
+        local_ref = weakref.ref(ObjectWithCyclicRef())
+
+        # Remote workers.
+        actors = [GarbageHolder.remote() for _ in range(2)]
+        assert local_ref() is not None
+        assert all(ray.get([a.has_garbage.remote() for a in actors]))
+
+        # GC should be triggered for all workers, including the local driver.
+        global_gc()
+        time.sleep(1)
+        assert local_ref() is None
+        assert not any(ray.get([a.has_garbage.remote() for a in actors]))
+    finally:
+        gc.enable()
+
+
 def test_local_refcounts(ray_start_regular):
     oid1 = ray.put(None)
     check_refcounts({oid1: (1, 0)})
diff --git a/src/ray/core_worker/core_worker.cc b/src/ray/core_worker/core_worker.cc
index 1d66982a1..8b421425b 100644
--- a/src/ray/core_worker/core_worker.cc
+++ b/src/ray/core_worker/core_worker.cc
@@ -68,12 +68,14 @@ CoreWorker::CoreWorker(const WorkerType worker_type, const Language language,
                        const std::string &log_dir, const std::string &node_ip_address,
                        int node_manager_port,
                        const TaskExecutionCallback &task_execution_callback,
-                       std::function<Status()> check_signals, bool ref_counting_enabled)
+                       std::function<Status()> check_signals,
+                       std::function<void()> gc_collect, bool ref_counting_enabled)
     : worker_type_(worker_type),
       language_(language),
       log_dir_(log_dir),
       ref_counting_enabled_(ref_counting_enabled),
       check_signals_(check_signals),
+      gc_collect_(gc_collect),
       worker_context_(worker_type, job_id),
       io_work_(io_service_),
       client_call_manager_(new rpc::ClientCallManager(io_service_)),
@@ -675,6 +677,18 @@ Status CoreWorker::Delete(const std::vector<ObjectID> &object_ids, bool local_on
   return Status::OK();
 }
 
+void CoreWorker::TriggerGlobalGC() {
+  auto status = local_raylet_client_->GlobalGC(
+      [](const Status &status, const rpc::GlobalGCReply &reply) {
+        if (!status.ok()) {
+          RAY_LOG(ERROR) << "Failed to send global GC request: " << status.ToString();
+        }
+      });
+  if (!status.ok()) {
+    RAY_LOG(ERROR) << "Failed to send global GC request: " << status.ToString();
+  }
+}
+
 std::string CoreWorker::MemoryUsageString() {
   // Currently only the Plasma store returns a debug string.
   return plasma_store_provider_->MemoryUsageString();
@@ -1350,6 +1364,18 @@ void CoreWorker::HandleGetCoreWorkerStats(const rpc::GetCoreWorkerStatsRequest &
   send_reply_callback(Status::OK(), nullptr, nullptr);
 }
 
+void CoreWorker::HandleLocalGC(const rpc::LocalGCRequest &request,
+                               rpc::LocalGCReply *reply,
+                               rpc::SendReplyCallback send_reply_callback) {
+  if (gc_collect_ != nullptr) {
+    gc_collect_();
+    send_reply_callback(Status::OK(), nullptr, nullptr);
+  } else {
+    send_reply_callback(Status::NotImplemented("GC callback not defined"), nullptr,
+                        nullptr);
+  }
+}
+
 void CoreWorker::YieldCurrentFiber(FiberEvent &event) {
   RAY_CHECK(worker_context_.CurrentActorIsAsync());
   boost::this_fiber::yield();
diff --git a/src/ray/core_worker/core_worker.h b/src/ray/core_worker/core_worker.h
index 203ebd2ce..e9249f2c9 100644
--- a/src/ray/core_worker/core_worker.h
+++ b/src/ray/core_worker/core_worker.h
@@ -76,6 +76,7 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
              const std::string &log_dir, const std::string &node_ip_address,
              int node_manager_port, const TaskExecutionCallback &task_execution_callback,
              std::function<Status()> check_signals = nullptr,
+             std::function<void()> gc_collect = nullptr,
              bool ref_counting_enabled = false);
 
   virtual ~CoreWorker();
@@ -280,6 +281,9 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
   Status Delete(const std::vector<ObjectID> &object_ids, bool local_only,
                 bool delete_creating_tasks);
 
+  /// Trigger garbage collection on each worker in the cluster.
+  void TriggerGlobalGC();
+
   /// Get a string describing object store memory usage for debugging purposes.
   ///
   /// \return std::string The string describing memory usage.
@@ -482,6 +486,10 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
                                 rpc::GetCoreWorkerStatsReply *reply,
                                 rpc::SendReplyCallback send_reply_callback) override;
 
+  /// Trigger local GC on this worker.
+  void HandleLocalGC(const rpc::LocalGCRequest &request, rpc::LocalGCReply *reply,
+                     rpc::SendReplyCallback send_reply_callback) override;
+
   ///
   /// Public methods related to async actor call. This should only be used when
   /// the actor is (1) direct actor and (2) using asyncio mode.
@@ -628,6 +636,11 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
   /// 1s) during long-running operations.
   std::function<Status()> check_signals_;
 
+  /// Application-language callback to trigger garbage collection in the language
+  /// runtime. This is required to free distributed references that may otherwise
+  /// be held up in garbage objects.
+  std::function<void()> gc_collect_;
+
   /// Shared state of the worker. Includes process-level and thread-level state.
   /// TODO(edoakes): we should move process-level state into this class and make
   /// this a ThreadContext.
diff --git a/src/ray/protobuf/core_worker.proto b/src/ray/protobuf/core_worker.proto
index 12c4dc383..2978909d6 100644
--- a/src/ray/protobuf/core_worker.proto
+++ b/src/ray/protobuf/core_worker.proto
@@ -218,6 +218,12 @@ message WaitForRefRemovedReply {
   repeated ObjectReferenceCount borrowed_refs = 1;
 }
 
+message LocalGCRequest {
+}
+
+message LocalGCReply {
+}
+
 service CoreWorkerService {
   // Push a task to a worker from the raylet.
   rpc AssignTask(AssignTaskRequest) returns (AssignTaskReply);
@@ -238,4 +244,6 @@ service CoreWorkerService {
   rpc GetCoreWorkerStats(GetCoreWorkerStatsRequest) returns (GetCoreWorkerStatsReply);
   // Wait for a borrower to finish using an object. Sent by the object's owner.
   rpc WaitForRefRemoved(WaitForRefRemovedRequest) returns (WaitForRefRemovedReply);
+  // Trigger local GC on the worker.
+  rpc LocalGC(LocalGCRequest) returns (LocalGCReply);
 }
diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto
index 6c936dc1f..b1891ca19 100644
--- a/src/ray/protobuf/gcs.proto
+++ b/src/ray/protobuf/gcs.proto
@@ -211,6 +211,8 @@ message HeartbeatTableData {
   repeated double resource_load_capacity = 7;
   // Object IDs that are in use by workers on this node manager's node.
   repeated bytes active_object_id = 8;
+  // Whether this node manager is requesting global GC.
+  bool should_global_gc = 9;
 }
 
 message HeartbeatBatchTableData {
diff --git a/src/ray/protobuf/node_manager.proto b/src/ray/protobuf/node_manager.proto
index 540389f99..d49c2d930 100644
--- a/src/ray/protobuf/node_manager.proto
+++ b/src/ray/protobuf/node_manager.proto
@@ -75,6 +75,12 @@ message GetNodeStatsReply {
   repeated TaskSpec ready_tasks = 5;
 }
 
+message GlobalGCRequest {
+}
+
+message GlobalGCReply {
+}
+
 // Service for inter-node-manager communication.
 service NodeManagerService {
   // Request a worker from the raylet.
@@ -87,4 +93,6 @@ service NodeManagerService {
   rpc PinObjectIDs(PinObjectIDsRequest) returns (PinObjectIDsReply);
   // Get the current node stats.
   rpc GetNodeStats(GetNodeStatsRequest) returns (GetNodeStatsReply);
+  // Trigger garbage collection in all workers across the cluster.
+  rpc GlobalGC(GlobalGCRequest) returns (GlobalGCReply);
 }
diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc
index fdb9ce5f1..afec1b964 100644
--- a/src/ray/raylet/node_manager.cc
+++ b/src/ray/raylet/node_manager.cc
@@ -304,6 +304,19 @@ void NodeManager::Heartbeat() {
     heartbeat_data->add_resource_load_capacity(resource_pair.second);
   }
 
+  // Set the global gc bit on the outgoing heartbeat message.
+  if (should_global_gc_) {
+    heartbeat_data->set_should_global_gc(true);
+    should_global_gc_ = false;
+  }
+
+  // Trigger local GC if needed. This throttles the frequency of local GC calls
+  // to at most once per heartbeat interval.
+  if (should_local_gc_) {
+    DoLocalGC();
+    should_local_gc_ = false;
+  }
+
   ray::Status status = gcs_client_->Nodes().AsyncReportHeartbeat(heartbeat_data,
                                                                  /*done*/ nullptr);
   RAY_CHECK_OK_PREPEND(status, "Heartbeat failed");
@@ -330,6 +343,26 @@ void NodeManager::Heartbeat() {
   });
 }
 
+void NodeManager::DoLocalGC() {
+  auto all_workers = worker_pool_.GetAllWorkers();
+  for (const auto &driver : worker_pool_.GetAllDrivers()) {
+    all_workers.push_back(driver);
+  }
+  RAY_LOG(WARNING) << "Sending local GC request to " << all_workers.size() << " workers.";
+  for (const auto &worker : all_workers) {
+    rpc::LocalGCRequest request;
+    auto status = worker->rpc_client()->LocalGC(
+        request, [](const ray::Status &status, const rpc::LocalGCReply &r) {
+          if (!status.ok()) {
+            RAY_LOG(ERROR) << "Failed to send local GC request: " << status.ToString();
+          }
+        });
+    if (!status.ok()) {
+      RAY_LOG(ERROR) << "Failed to send local GC request: " << status.ToString();
+    }
+  }
+}
+
 // TODO(edoakes): this function is problematic because it both sends warnings spuriously
 // under normal conditions and sometimes doesn't send a warning under actual deadlock
 // conditions. The current logic is to push a warning when: all running tasks are
@@ -650,6 +683,12 @@ void NodeManager::HeartbeatAdded(const ClientID &client_id,
                   << client_id;
     return;
   }
+
+  // Trigger local GC at the next heartbeat interval.
+  if (heartbeat_data.should_global_gc()) {
+    should_local_gc_ = true;
+  }
+
   SchedulingResources &remote_resources = it->second;
 
   ResourceSet remote_total(VectorFromProtobuf(heartbeat_data.resources_total_label()),
@@ -3272,6 +3311,15 @@ void NodeManager::HandleGetNodeStats(const rpc::GetNodeStatsRequest &request,
   }
 }
 
+void NodeManager::HandleGlobalGC(const rpc::GlobalGCRequest &request,
+                                 rpc::GlobalGCReply *reply,
+                                 rpc::SendReplyCallback send_reply_callback) {
+  RAY_LOG(WARNING) << "Broadcasting global GC request to all raylets.";
+  should_global_gc_ = true;
+  // We won't see our own request, so trigger local GC in the next heartbeat.
+  should_local_gc_ = true;
+}
+
 void NodeManager::RecordMetrics() {
   recorded_metrics_ = true;
   if (stats::StatsConfig::instance().IsStatsDisabled()) {
diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h
index f3b86dabe..5788eee8d 100644
--- a/src/ray/raylet/node_manager.h
+++ b/src/ray/raylet/node_manager.h
@@ -563,6 +563,13 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
                           rpc::GetNodeStatsReply *reply,
                           rpc::SendReplyCallback send_reply_callback) override;
 
+  /// Handle a `GlobalGC` request.
+  void HandleGlobalGC(const rpc::GlobalGCRequest &request, rpc::GlobalGCReply *reply,
+                      rpc::SendReplyCallback send_reply_callback) override;
+
+  /// Trigger local GC on each worker of this raylet.
+  void DoLocalGC();
+
   /// Push an error to the driver if this node is full of actors and so we are
   /// unable to schedule new tasks or actors at all.
   void WarnResourceDeadlock();
@@ -669,6 +676,14 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
   /// Whether new schedule is enabled.
   const bool new_scheduler_enabled_;
 
+  /// Whether to trigger global GC in the next heartbeat. This will broadcast
+  /// a global GC message to all raylets except for this one.
+  bool should_global_gc_ = false;
+
+  /// Whether to trigger local GC in the next heartbeat. This will trigger gc
+  /// on all local workers of this raylet.
+  bool should_local_gc_ = false;
+
   /// The new resource scheduler for direct task calls.
   std::shared_ptr<ClusterResourceScheduler> new_resource_scheduler_;
   /// Map of leased workers to their current resource usage.
diff --git a/src/ray/raylet/raylet_client.cc b/src/ray/raylet/raylet_client.cc
index 022b1ea11..9e45a3b54 100644
--- a/src/ray/raylet/raylet_client.cc
+++ b/src/ray/raylet/raylet_client.cc
@@ -377,4 +377,10 @@ Status raylet::RayletClient::PinObjectIDs(
   return grpc_client_->PinObjectIDs(request, callback);
 }
 
+Status raylet::RayletClient::GlobalGC(
+    const rpc::ClientCallback<rpc::GlobalGCReply> &callback) {
+  rpc::GlobalGCRequest request;
+  return grpc_client_->GlobalGC(request, callback);
+}
+
 }  // namespace ray
diff --git a/src/ray/raylet/raylet_client.h b/src/ray/raylet/raylet_client.h
index 8ef8764c4..f1716bf52 100644
--- a/src/ray/raylet/raylet_client.h
+++ b/src/ray/raylet/raylet_client.h
@@ -257,6 +257,8 @@ class RayletClient : public WorkerLeaseInterface {
       const rpc::Address &caller_address, const std::vector<ObjectID> &object_ids,
       const ray::rpc::ClientCallback<ray::rpc::PinObjectIDsReply> &callback);
 
+  ray::Status GlobalGC(const rpc::ClientCallback<rpc::GlobalGCReply> &callback);
+
   WorkerID GetWorkerID() const { return worker_id_; }
 
   JobID GetJobID() const { return job_id_; }
diff --git a/src/ray/rpc/node_manager/node_manager_client.h b/src/ray/rpc/node_manager/node_manager_client.h
index 93e438148..b07df522c 100644
--- a/src/ray/rpc/node_manager/node_manager_client.h
+++ b/src/ray/rpc/node_manager/node_manager_client.h
@@ -76,6 +76,9 @@ class NodeManagerWorkerClient
   /// Notify the raylet to pin the provided object IDs.
   RPC_CLIENT_METHOD(NodeManagerService, PinObjectIDs, grpc_client_, )
 
+  /// Trigger global GC across the cluster.
+  RPC_CLIENT_METHOD(NodeManagerService, GlobalGC, grpc_client_, )
+
  private:
   /// Constructor.
   ///
diff --git a/src/ray/rpc/node_manager/node_manager_server.h b/src/ray/rpc/node_manager/node_manager_server.h
index bcb2328de..c73a3b420 100644
--- a/src/ray/rpc/node_manager/node_manager_server.h
+++ b/src/ray/rpc/node_manager/node_manager_server.h
@@ -15,7 +15,8 @@ namespace rpc {
   RPC_SERVICE_HANDLER(NodeManagerService, ReturnWorker, 100)       \
   RPC_SERVICE_HANDLER(NodeManagerService, ForwardTask, 100)        \
   RPC_SERVICE_HANDLER(NodeManagerService, PinObjectIDs, 100)       \
-  RPC_SERVICE_HANDLER(NodeManagerService, GetNodeStats, 1)
+  RPC_SERVICE_HANDLER(NodeManagerService, GetNodeStats, 1)         \
+  RPC_SERVICE_HANDLER(NodeManagerService, GlobalGC, 1)
 
 /// Interface of the `NodeManagerService`, see `src/ray/protobuf/node_manager.proto`.
 class NodeManagerServiceHandler {
@@ -50,6 +51,9 @@ class NodeManagerServiceHandler {
   virtual void HandleGetNodeStats(const GetNodeStatsRequest &request,
                                   GetNodeStatsReply *reply,
                                   SendReplyCallback send_reply_callback) = 0;
+
+  virtual void HandleGlobalGC(const GlobalGCRequest &request, GlobalGCReply *reply,
+                              SendReplyCallback send_reply_callback) = 0;
 };
 
 /// The `GrpcService` for `NodeManagerService`.
diff --git a/src/ray/rpc/worker/core_worker_client.h b/src/ray/rpc/worker/core_worker_client.h
index abfc0ef5d..5ace160de 100644
--- a/src/ray/rpc/worker/core_worker_client.h
+++ b/src/ray/rpc/worker/core_worker_client.h
@@ -149,6 +149,11 @@ class CoreWorkerClientInterface {
     return Status::NotImplemented("");
   }
 
+  virtual ray::Status LocalGC(const LocalGCRequest &request,
+                              const ClientCallback<LocalGCReply> &callback) {
+    return Status::NotImplemented("");
+  }
+
   virtual ray::Status WaitForRefRemoved(
       const WaitForRefRemovedRequest &request,
       const ClientCallback<WaitForRefRemovedReply> &callback) {
@@ -189,6 +194,8 @@ class CoreWorkerClient : public std::enable_shared_from_this<CoreWorkerClient>,
 
   RPC_CLIENT_METHOD(CoreWorkerService, GetCoreWorkerStats, grpc_client_, override)
 
+  RPC_CLIENT_METHOD(CoreWorkerService, LocalGC, grpc_client_, override)
+
   RPC_CLIENT_METHOD(CoreWorkerService, WaitForRefRemoved, grpc_client_, override)
 
   ray::Status PushActorTask(std::unique_ptr<PushTaskRequest> request,
diff --git a/src/ray/rpc/worker/core_worker_server.h b/src/ray/rpc/worker/core_worker_server.h
index e32660b63..7d384d213 100644
--- a/src/ray/rpc/worker/core_worker_server.h
+++ b/src/ray/rpc/worker/core_worker_server.h
@@ -22,7 +22,8 @@ namespace rpc {
   RPC_SERVICE_HANDLER(CoreWorkerService, WaitForObjectEviction, 9999)         \
   RPC_SERVICE_HANDLER(CoreWorkerService, WaitForRefRemoved, 9999)             \
   RPC_SERVICE_HANDLER(CoreWorkerService, KillActor, 9999)                     \
-  RPC_SERVICE_HANDLER(CoreWorkerService, GetCoreWorkerStats, 100)
+  RPC_SERVICE_HANDLER(CoreWorkerService, GetCoreWorkerStats, 100)             \
+  RPC_SERVICE_HANDLER(CoreWorkerService, LocalGC, 100)
 
 #define RAY_CORE_WORKER_DECLARE_RPC_HANDLERS                              \
   DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(AssignTask)                     \
@@ -32,7 +33,8 @@ namespace rpc {
   DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(WaitForObjectEviction)          \
   DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(WaitForRefRemoved)              \
   DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(KillActor)                      \
-  DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(GetCoreWorkerStats)
+  DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(GetCoreWorkerStats)             \
+  DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(LocalGC)
 
 /// Interface of the `CoreWorkerServiceHandler`, see `src/ray/protobuf/core_worker.proto`.
 class CoreWorkerServiceHandler {