Clean up event loop and callbacks when processes exit. (#1125)

* Clean up event loop and callbacks when processes exit.

* Fix bug.

src/global_scheduler/global_scheduler.cc

@@ -121,6 +121,7 @@ GlobalSchedulerState *GlobalSchedulerState_init(event_loop *loop,
                                                 const char *redis_primary_addr,
                                                 int redis_primary_port) {
   GlobalSchedulerState *state = new GlobalSchedulerState();
+  state->loop = loop;
   state->db = db_connect(std::string(redis_primary_addr), redis_primary_port,
                          "global_scheduler", node_ip_address, 0, NULL);
   db_attach(state->db, loop, false);
@@ -152,6 +153,12 @@ void GlobalSchedulerState_free(GlobalSchedulerState *state) {
     Task_free(pending_task);
   }
   state->pending_tasks.clear();
+
+  /* Destroy the event loop. */
+  destroy_outstanding_callbacks(state->loop);
+  event_loop_destroy(state->loop);
+  state->loop = NULL;
+
   /* Free the global scheduler state. */
   delete state;
 }

src/local_scheduler/local_scheduler.cc

@@ -197,7 +197,9 @@ void LocalSchedulerState_free(LocalSchedulerState *state) {
   /* Free the algorithm state. */
   SchedulingAlgorithmState_free(state->algorithm_state);
   state->algorithm_state = NULL;
+
   /* Destroy the event loop. */
+  destroy_outstanding_callbacks(state->loop);
   event_loop_destroy(state->loop);
   state->loop = NULL;
 

src/plasma/plasma_manager.cc

@@ -517,8 +517,12 @@ void PlasmaManagerState_free(PlasmaManagerState *state) {
 
   ARROW_CHECK_OK(state->plasma_conn->Disconnect());
   delete state->plasma_conn;
+
+  /* Destroy the event loop. */
   destroy_outstanding_callbacks(state->loop);
   event_loop_destroy(state->loop);
+  state->loop = NULL;
+
   delete state;
 }