PYTHON-1332 - Update session's last_use

Author: ajdavis

pymongo/bulk.py

@@ -320,9 +320,9 @@ def execute_command(self, sock_info, generator, write_concern, session):
  if self.bypass_doc_val and sock_info.max_wire_version >= 4:
  cmd['bypassDocumentValidation'] = True
  if s:
- cmd['lsid'] = s.session_id
+ cmd['lsid'] = s._use_lsid()
  bwc = _BulkWriteContext(db_name, cmd, sock_info, op_id,
- listeners)
+ listeners, s)
 
  results = _do_batched_write_command(
  self.namespace, run.op_type, cmd,

pymongo/client_session.py

@@ -120,6 +120,13 @@ def has_ended(self):
  """True if this session is finished."""
  return self._server_session is None
 
+ def _use_lsid(self):
+ # Internal function.
+ if self._server_session is None:
+ raise InvalidOperation("Cannot use ended session")
+
+ return self._server_session.use_lsid()
+
 
 class _ServerSession(object):
  def __init__(self):
@@ -133,6 +140,10 @@ def timed_out(self, session_timeout_minutes):
  # Timed out if we have less than a minute to live.
  return idle_seconds > (session_timeout_minutes - 1) * 60
 
+ def use_lsid(self):
+ self.last_use = monotonic.time()
+ return self.session_id
+
 
 class _ServerSessionPool(collections.deque):
  """Pool of _ServerSession objects.

pymongo/collection.py

@@ -635,12 +635,12 @@ def gen():
  command['bypassDocumentValidation'] = True
  bwc = message._BulkWriteContext(
  self.database.name, command, sock_info, op_id,
- self.database.client._event_listeners)
+ self.database.client._event_listeners, session=None)
  if acknowledged:
  # Batched insert command.
  with self.__database.client._tmp_session(session) as s:
  if s:
- command['lsid'] = s.session_id
+ command['lsid'] = s._use_lsid()
  results = message._do_batched_write_command(
  self.database.name + ".$cmd", message._INSERT, command,
  gen(), check_keys, self.__write_response_codec_options, bwc)

pymongo/message.py

@@ -174,7 +174,7 @@ def _gen_explain_command(
  explain = SON([('explain', cmd)])
 
  if session:
- explain['lsid'] = session.session_id
+ explain['lsid'] = session._use_lsid()
 
  return explain
 
@@ -213,7 +213,7 @@ def _gen_find_command(coll, spec, projection, skip, limit, batch_size, options,
  for opt, val in _OPTIONS.items()
  if options & val])
  if session:
- cmd['lsid'] = session.session_id
+ cmd['lsid'] = session._use_lsid()
  return cmd
 
 
@@ -227,7 +227,7 @@ def _gen_get_more_command(cursor_id, coll, batch_size, max_await_time_ms,
  if max_await_time_ms is not None:
  cmd['maxTimeMS'] = max_await_time_ms
  if session:
- cmd['lsid'] = session.session_id
+ cmd['lsid'] = session._use_lsid()
  return cmd
 
 
@@ -565,10 +565,11 @@ class _BulkWriteContext(object):
  """A wrapper around SocketInfo for use with write splitting functions."""
 
  __slots__ = ('db_name', 'command', 'sock_info', 'op_id',
- 'name', 'field', 'publish', 'start_time', 'listeners')
+ 'name', 'field', 'publish', 'start_time', 'listeners',
+ 'session')
 
- def __init__(
- self, database_name, command, sock_info, operation_id, listeners):
+ def __init__(self, database_name, command, sock_info, operation_id,
+   listeners, session):
  self.db_name = database_name
  self.command = command
  self.sock_info = sock_info
@@ -578,6 +579,7 @@ def __init__(
  self.name = next(iter(command))
  self.field = _FIELD_MAP[self.name]
  self.start_time = datetime.datetime.now() if self.publish else None
+ self.session = session
 
  @property
  def max_bson_size(self):
@@ -628,6 +630,9 @@ def legacy_write(self, request_id, msg, max_doc_size, acknowledged, docs):
  def write_command(self, request_id, msg, docs):
  """A proxy for SocketInfo.write_command that handles event publishing.
  """
+ if self.session:
+ # Update last_use time.
+ self.session._use_lsid()
  if self.publish:
  duration = datetime.datetime.now() - self.start_time
  self._start(request_id, docs)

pymongo/network.py

@@ -82,7 +82,7 @@ def command(sock, dbname, spec, slave_ok, is_mongos,
  if spec.__class__ is dict:
  # Ensure command name remains in first place.
  spec = SON(spec)
- spec['lsid'] = session.session_id
+ spec['lsid'] = session._use_lsid()
 
  # Publish the original command document, perhaps with session id.
  orig = spec

test/test_session.py

@@ -24,6 +24,7 @@
 from pymongo.errors import (ConfigurationError,
  InvalidOperation,
  OperationFailure)
+from pymongo.monotonic import time as _time
 from test import IntegrationTest, client_context, db_user, db_pwd
 from test.utils import ignore_deprecations, rs_or_single_client, EventListener
 
@@ -74,12 +75,16 @@ def _test_ops(self, client, *ops, **kwargs):
 
  for f, args, kw in ops:
  with client.start_session() as s:
+ last_use = s._server_session.last_use
+ start = _time()
+ self.assertLessEqual(last_use, start)
  listener.results.clear()
  # In case "f" modifies its inputs.
  args = copy.copy(args)
  kw = copy.copy(kw)
  kw['session'] = s
  f(*args, **kw)
+ self.assertGreaterEqual(s._server_session.last_use, start)
  self.assertGreaterEqual(len(listener.results['started']), 1)
  for event in listener.results['started']:
  self.assertTrue(