Remove unnecessary db request while adding task
Using the timestamp assigned by the coordinator node does not
provide any benefit over using a client-provided timestamp.
@@ -276,6 +276,12 @@ class PullQueue(Queue):
if not hasattr(task, 'payloadBase64'):
raise InvalidTaskInfo('{} is missing a payload.'.format(task))
+ enqueue_time = datetime.datetime.utcnow()
+ try:
+ lease_expires = task.leaseTimestamp
+ except AttributeError:
+ lease_expires = datetime.datetime.utcfromtimestamp(0)
+
insert_task = SimpleStatement("""
INSERT INTO pull_queue_tasks (
app, queue, id, payload,
@@ -283,7 +289,7 @@ class PullQueue(Queue):
)
VALUES (
%(app)s, %(queue)s, %(id)s, %(payload)s,
- dateof(now()), %(lease_expires)s, 0, %(tag)s
+ %(enqueued)s, %(lease_expires)s, 0, %(tag)s
)
IF NOT EXISTS
""", retry_policy=BASIC_RETRIES)
@@ -291,7 +297,9 @@ class PullQueue(Queue):
'app': self.app,
'queue': self.name,
'id': task.id,
- 'payload': task.payloadBase64
+ 'payload': task.payloadBase64,
+ 'enqueued': enqueue_time,
+ 'lease_expires': lease_expires
}
try:
@@ -299,25 +307,13 @@ class PullQueue(Queue):
except AttributeError:
parameters['tag'] = None
- try:
- parameters['lease_expires'] = task.leaseTimestamp
- except AttributeError:
- parameters['lease_expires'] = 0
-
result = self.db_access.session.execute(insert_task, parameters)[0]
if not result.applied:
raise InvalidTaskInfo('Task name already taken: {}'.format(task.id))
- # Retrieve the date values that Cassandra generated.
- select_task = """
- SELECT enqueued, lease_expires FROM pull_queue_tasks
- WHERE app = %(app)s AND queue = %(queue)s AND id = %(id)s
- """
- parameters = {'app': self.app, 'queue': self.name, 'id': task.id}
- response = self.db_access.session.execute(select_task, parameters)[0]
task.queueName = self.name
- task.enqueueTimestamp = response.enqueued
- task.leaseTimestamp = response.lease_expires
+ task.enqueueTimestamp = enqueue_time
+ task.leaseTimestamp = lease_expires
# Create an index entry so the task can be queried by ETA. This can't be
# done in a batch because the payload from the previous insert can be up
