diff --git a/gstlal-burst/bin/gstlal_feature_synchronizer b/gstlal-burst/bin/gstlal_feature_synchronizer index caef73d054b09f5c899a2c775613681cba5c181e..da41740e73de1b7924d0030c155990091386f085 100755 --- a/gstlal-burst/bin/gstlal_feature_synchronizer +++ b/gstlal-burst/bin/gstlal_feature_synchronizer @@ -56,6 +56,7 @@ def parse_command_line(): parser.add_option("--processing-cadence", type = "float", default = 0.1, help = "Rate at which the synchronizer acquires and processes data. Default = 0.1 seconds.") parser.add_option("--request-timeout", type = "float", default = 0.2, help = "Timeout for requesting messages from a topic. Default = 0.2 seconds.") parser.add_option("--latency-timeout", type = "float", default = 5, help = "Maximum time before incoming data is dropped for a given timestamp. Default = 5 seconds.") + parser.add_option("--sample-rate", type = "int", metavar = "Hz", default = 1, help = "Set the sample rate for feature timeseries output, must be a power of 2. Default = 1 Hz.") parser.add_option("--no-drop", default=False, action="store_true", help = "If set, do not drop incoming features based on the latency timeout. Default = False.") parser.add_option("--kafka-server", metavar = "string", help = "Sets the server url that the kafka topic is hosted on. Required.") parser.add_option("--input-topic-basename", metavar = "string", help = "Sets the input kafka topic basename, i.e. {basename}_%02d. Required.") @@ -82,6 +83,7 @@ class StreamSynchronizer(object): self.processing_cadence = options.processing_cadence self.request_timeout = options.request_timeout self.latency_timeout = options.latency_timeout + self.sample_rate = options.sample_rate self.no_drop = options.no_drop self.is_running = False @@ -104,8 +106,8 @@ class StreamSynchronizer(object): ### initialize queues self.last_timestamp = 0 - # 1 minute queue for incoming buffers - self.feature_queue = PriorityQueue(maxsize = 60 * self.num_topics) + # 30 second queue for incoming buffers + self.feature_queue = PriorityQueue(maxsize = 30 * self.sample_rate * self.num_topics) # 5 minute queue for outgoing buffers self.feature_buffer = deque(maxlen = 300) diff --git a/gstlal-burst/bin/gstlal_ll_feature_extractor_pipe b/gstlal-burst/bin/gstlal_ll_feature_extractor_pipe index a0cfbad7899742cd46fa6ab67bc0da9cc5eb14c5..cd085faf487b738f4b617b6e8c605d56eabab3c3 100755 --- a/gstlal-burst/bin/gstlal_ll_feature_extractor_pipe +++ b/gstlal-burst/bin/gstlal_ll_feature_extractor_pipe @@ -327,6 +327,7 @@ if options.save_format == 'kafka': synchronizer_options = { "latency-timeout": options.latency_timeout, + "sample-rate": options.sample_rate, "input-topic-basename": options.kafka_topic, "output-topic-basename": '_'.join(['synchronizer', options.tag]) }