From d9b152a44d6451f452cd2ae1d12c5991c63d9cf4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philip=20Dub=C3=A9?= Date: Thu, 7 Dec 2023 18:45:47 +0000 Subject: [PATCH] Replace `chan bool` with `chan struct{}` (#777) Slight perf optimization that also clearly indicates that the value given to channel does not mean anything --- flow/activities/flowable.go | 12 ++++++------ flow/activities/snapshot_activity.go | 2 +- flow/connectors/bigquery/qrep_avro_sync.go | 2 +- flow/connectors/eventhub/eventhub.go | 2 +- flow/connectors/postgres/cdc.go | 2 +- flow/connectors/postgres/postgres_repl_test.go | 2 +- flow/connectors/postgres/qrep_query_executor.go | 2 +- flow/connectors/postgres/slot_signal.go | 4 ++-- flow/connectors/snowflake/qrep_avro_sync.go | 2 +- flow/connectors/utils/avro/avro_writer.go | 2 +- flow/connectors/utils/heartbeat.go | 4 ++-- 11 files changed, 18 insertions(+), 18 deletions(-) diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go index 6b71e6b90..2fc36da69 100644 --- a/flow/activities/flowable.go +++ b/flow/activities/flowable.go @@ -266,7 +266,7 @@ func (a *FlowableActivity) StartFlow(ctx context.Context, }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() syncStartTime := time.Now() @@ -364,7 +364,7 @@ func (a *FlowableActivity) StartNormalize( return fmt.Sprintf("normalizing records from batch for job - %s", input.FlowConnectionConfigs.FlowJobName) }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() log.Info("initializing table schema...") @@ -443,7 +443,7 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context, }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() partitions, err := srcConn.GetQRepPartitions(config, last) @@ -574,7 +574,7 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context, }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() res, err := dstConn.SyncQRepRecords(config, partition, stream) @@ -618,7 +618,7 @@ func (a *FlowableActivity) ConsolidateQRepPartitions(ctx context.Context, config }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() err = dstConn.ConsolidateQRepPartitions(config) @@ -919,7 +919,7 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context, }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() res, err := dstConn.SyncQRepRecords(config, partition, stream) diff --git a/flow/activities/snapshot_activity.go b/flow/activities/snapshot_activity.go index a49067427..98a0025d5 100644 --- a/flow/activities/snapshot_activity.go +++ b/flow/activities/snapshot_activity.go @@ -21,7 +21,7 @@ func (a *SnapshotActivity) CloseSlotKeepAlive(flowJobName string) error { } if s, ok := a.SnapshotConnections[flowJobName]; ok { - s.signal.CloneComplete <- true + s.signal.CloneComplete <- struct{}{} s.connector.Close() } diff --git a/flow/connectors/bigquery/qrep_avro_sync.go b/flow/connectors/bigquery/qrep_avro_sync.go index 5f395f358..c31442ed2 100644 --- a/flow/connectors/bigquery/qrep_avro_sync.go +++ b/flow/connectors/bigquery/qrep_avro_sync.go @@ -325,7 +325,7 @@ func (s *QRepAvroSyncMethod) writeToStage( }, ) defer func() { - shutdown <- true + shutdown <- struct{}{} }() var avroFile *avro.AvroFile diff --git a/flow/connectors/eventhub/eventhub.go b/flow/connectors/eventhub/eventhub.go index 961fb979f..896b7fd9a 100644 --- a/flow/connectors/eventhub/eventhub.go +++ b/flow/connectors/eventhub/eventhub.go @@ -227,7 +227,7 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S ) }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() // if env var PEERDB_BETA_EVENTHUB_PUSH_ASYNC=true diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go index fc0063c1a..f0ef60f1b 100644 --- a/flow/connectors/postgres/cdc.go +++ b/flow/connectors/postgres/cdc.go @@ -209,7 +209,7 @@ func (p *PostgresCDCSource) consumeStream( }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() standbyMessageTimeout := req.IdleTimeout diff --git a/flow/connectors/postgres/postgres_repl_test.go b/flow/connectors/postgres/postgres_repl_test.go index 7e10d9a15..d3831fee6 100644 --- a/flow/connectors/postgres/postgres_repl_test.go +++ b/flow/connectors/postgres/postgres_repl_test.go @@ -127,7 +127,7 @@ func (suite *PostgresReplicationSnapshotTestSuite) TestSimpleSlotCreation() { log.Infof("signaling clone complete for %s after waiting for 2 seconds", flowJobName) time.Sleep(2 * time.Second) - signal.CloneComplete <- true + signal.CloneComplete <- struct{}{} log.Infof("successfully setup replication for %s", flowJobName) } diff --git a/flow/connectors/postgres/qrep_query_executor.go b/flow/connectors/postgres/qrep_query_executor.go index 85c0fb2cc..c8ea1bcad 100644 --- a/flow/connectors/postgres/qrep_query_executor.go +++ b/flow/connectors/postgres/qrep_query_executor.go @@ -83,7 +83,7 @@ func (qe *QRepQueryExecutor) executeQueryInTx(tx pgx.Tx, cursorName string, fetc }) defer func() { - shutdownCh <- true + shutdownCh <- struct{}{} }() } diff --git a/flow/connectors/postgres/slot_signal.go b/flow/connectors/postgres/slot_signal.go index e55704983..966057559 100644 --- a/flow/connectors/postgres/slot_signal.go +++ b/flow/connectors/postgres/slot_signal.go @@ -11,13 +11,13 @@ type SlotCreationResult struct { // 2. CloneComplete - which can be waited on to ensure that the clone has completed. type SlotSignal struct { SlotCreated chan *SlotCreationResult - CloneComplete chan bool + CloneComplete chan struct{} } // NewSlotSignal returns a new SlotSignal. func NewSlotSignal() *SlotSignal { return &SlotSignal{ SlotCreated: make(chan *SlotCreationResult, 1), - CloneComplete: make(chan bool, 1), + CloneComplete: make(chan struct{}, 1), } } diff --git a/flow/connectors/snowflake/qrep_avro_sync.go b/flow/connectors/snowflake/qrep_avro_sync.go index 920e01d1c..59259e3a5 100644 --- a/flow/connectors/snowflake/qrep_avro_sync.go +++ b/flow/connectors/snowflake/qrep_avro_sync.go @@ -318,7 +318,7 @@ func (s *SnowflakeAvroSyncMethod) putFileToStage(avroFile *avro.AvroFile, stage }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() if _, err := s.connector.database.Exec(putCmd); err != nil { diff --git a/flow/connectors/utils/avro/avro_writer.go b/flow/connectors/utils/avro/avro_writer.go index 3c7fa5830..1bb65d431 100644 --- a/flow/connectors/utils/avro/avro_writer.go +++ b/flow/connectors/utils/avro/avro_writer.go @@ -135,7 +135,7 @@ func (p *peerDBOCFWriter) writeRecordsToOCFWriter(ocfWriter *goavro.OCFWriter) ( }) defer func() { - shutdown <- true + shutdown <- struct{}{} }() } diff --git a/flow/connectors/utils/heartbeat.go b/flow/connectors/utils/heartbeat.go index b16735b6b..164c64e3c 100644 --- a/flow/connectors/utils/heartbeat.go +++ b/flow/connectors/utils/heartbeat.go @@ -13,9 +13,9 @@ func HeartbeatRoutine( ctx context.Context, interval time.Duration, message func() string, -) chan bool { +) chan struct{} { counter := 1 - shutdown := make(chan bool) + shutdown := make(chan struct{}) go func() { for { msg := fmt.Sprintf("heartbeat #%d: %s", counter, message())