Merge pull request #1014

Author: alcaeus

php_phongo.c

@@ -604,7 +604,7 @@ bool phongo_parse_read_preference(zval* options, zval** zreadPreference TSRMLS_D
  * not NULL, the option will be appended. If zsession is not NULL, it will be
  * assigned to the option. On error, false is returned and an exception is
  * thrown. */
-static bool phongo_parse_session(zval* options, mongoc_client_t* client, bson_t* mongoc_opts, zval** zsession TSRMLS_DC) /* {{{ */
+bool phongo_parse_session(zval* options, mongoc_client_t* client, bson_t* mongoc_opts, zval** zsession TSRMLS_DC) /* {{{ */
 {
 	zval*                          option = NULL;
 	const mongoc_client_session_t* client_session;

php_phongo.h

@@ -147,6 +147,7 @@ const mongoc_write_concern_t* phongo_write_concern_from_zval(zval* zwrite_concer
 php_phongo_server_description_type_t php_phongo_server_description_type(mongoc_server_description_t* sd);
 
 bool phongo_parse_read_preference(zval* options, zval** zreadPreference TSRMLS_DC);
+bool phongo_parse_session(zval* options, mongoc_client_t* client, bson_t* mongoc_opts, zval** zsession TSRMLS_DC);
 
 zval* php_phongo_prep_legacy_option(zval* options, const char* key, bool* allocated TSRMLS_DC);
 void  php_phongo_prep_legacy_option_free(zval* options TSRMLS_DC);

scripts/presets/travis/sharded_clusters/cluster.json

@@ -69,6 +69,13 @@
             "logappend": true,
             "port": 4300,
             "bind_ip_all": true
+        },
+        {
+            "logpath": "/tmp/MO/SHARDED/ROUTER/4301/mongod.log",
+            "ipv6": true,
+            "logappend": true,
+            "port": 4301,
+            "bind_ip_all": true
         }
     ]
 }

scripts/presets/travis/sharded_clusters/cluster_replset.json

@@ -91,6 +91,13 @@
             "logappend": true,
             "port": 4430,
             "bind_ip_all": true
+        },
+        {
+            "logpath": "/tmp/MO/SHARDED-RS/ROUTER/4431/mongod.log",
+            "ipv6": true,
+            "logappend": true,
+            "port": 4431,
+            "bind_ip_all": true
         }
     ]
 }

src/LIBMONGOC_VERSION_CURRENT

@@ -1 +1 @@
-1.15.0
+1.15.1-20190828+gitcf50d6315a

src/MongoDB/Manager.c

@@ -254,15 +254,28 @@ static void php_phongo_manager_prep_uri_options(zval* options TSRMLS_DC) /* {{{
 /* Selects a server for an execute method. If "for_writes" is true, a primary
  * will be selected. Otherwise, a read preference will be used to select the
  * server. If zreadPreference is NULL, the client's read preference will be
- * used.
+ * used. If zsession is a session object in a sharded transaction, the session
+ * will be checked whether it is pinned to a server. If so, that server will be
+ * selected. Otherwise, server selection
  *
  * On success, server_id will be set and the function will return true;
  * otherwise, false is returned and an exception is thrown. */
-static bool php_phongo_manager_select_server(bool for_writes, zval* zreadPreference, mongoc_client_t* client, uint32_t* server_id TSRMLS_DC) /* {{{ */
+static bool php_phongo_manager_select_server(bool for_writes, zval* zreadPreference, zval* zsession, mongoc_client_t* client, uint32_t* server_id TSRMLS_DC) /* {{{ */
 {
-	const mongoc_read_prefs_t*   read_preference = NULL;
 	mongoc_server_description_t* selected_server;
-	bson_error_t                 error = { 0 };
+	const mongoc_read_prefs_t*   read_preference = NULL;
+	bson_error_t                 error           = { 0 };
+
+	if (zsession) {
+		const mongoc_client_session_t* session = Z_SESSION_OBJ_P(zsession)->client_session;
+
+		/* Attempt to fetch server pinned to session */
+		if (mongoc_client_session_get_server_id(session) > 0) {
+			*server_id = mongoc_client_session_get_server_id(session);
+
+			return true;
+		}
+	}
 
 	if (!for_writes) {
 		read_preference = zreadPreference ? phongo_read_preference_from_zval(zreadPreference TSRMLS_CC) : mongoc_client_get_read_prefs(client);
@@ -336,6 +349,7 @@ static PHP_METHOD(Manager, executeCommand)
 	zval*                 options         = NULL;
 	bool                  free_options    = false;
 	zval*                 zreadPreference = NULL;
+	zval*                 zsession        = NULL;
 	uint32_t              server_id       = 0;
 	DECLARE_RETURN_VALUE_USED
 
@@ -347,12 +361,17 @@ static PHP_METHOD(Manager, executeCommand)
 
 	options = php_phongo_prep_legacy_option(options, "readPreference", &free_options TSRMLS_CC);
 
+	if (!phongo_parse_session(options, intern->client, NULL, &zsession TSRMLS_CC)) {
+		/* Exception should already have been thrown */
+		goto cleanup;
+	}
+
 	if (!phongo_parse_read_preference(options, &zreadPreference TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		goto cleanup;
 	}
 
-	if (!php_phongo_manager_select_server(false, zreadPreference, intern->client, &server_id TSRMLS_CC)) {
+	if (!php_phongo_manager_select_server(false, zreadPreference, zsession, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		goto cleanup;
 	}
@@ -376,6 +395,7 @@ static PHP_METHOD(Manager, executeReadCommand)
 	zval*                 options         = NULL;
 	zval*                 zreadPreference = NULL;
 	uint32_t              server_id       = 0;
+	zval*                 zsession        = NULL;
 	DECLARE_RETURN_VALUE_USED
 
 	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sO|a!", &db, &db_len, &command, php_phongo_command_ce, &options) == FAILURE) {
@@ -384,12 +404,17 @@ static PHP_METHOD(Manager, executeReadCommand)
 
 	intern = Z_MANAGER_OBJ_P(getThis());
 
+	if (!phongo_parse_session(options, intern->client, NULL, &zsession TSRMLS_CC)) {
+		/* Exception should already have been thrown */
+		return;
+	}
+
 	if (!phongo_parse_read_preference(options, &zreadPreference TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		return;
 	}
 
-	if (!php_phongo_manager_select_server(false, zreadPreference, intern->client, &server_id TSRMLS_CC)) {
+	if (!php_phongo_manager_select_server(false, zreadPreference, zsession, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		return;
 	}
@@ -407,6 +432,7 @@ static PHP_METHOD(Manager, executeWriteCommand)
 	zval*                 command;
 	zval*                 options   = NULL;
 	uint32_t              server_id = 0;
+	zval*                 zsession  = NULL;
 	DECLARE_RETURN_VALUE_USED
 
 	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sO|a!", &db, &db_len, &command, php_phongo_command_ce, &options) == FAILURE) {
@@ -415,7 +441,12 @@ static PHP_METHOD(Manager, executeWriteCommand)
 
 	intern = Z_MANAGER_OBJ_P(getThis());
 
-	if (!php_phongo_manager_select_server(true, NULL, intern->client, &server_id TSRMLS_CC)) {
+	if (!phongo_parse_session(options, intern->client, NULL, &zsession TSRMLS_CC)) {
+		/* Exception should already have been thrown */
+		return;
+	}
+
+	if (!php_phongo_manager_select_server(true, NULL, zsession, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		return;
 	}
@@ -433,6 +464,7 @@ static PHP_METHOD(Manager, executeReadWriteCommand)
 	zval*                 command;
 	zval*                 options   = NULL;
 	uint32_t              server_id = 0;
+	zval*                 zsession  = NULL;
 	DECLARE_RETURN_VALUE_USED
 
 	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sO|a!", &db, &db_len, &command, php_phongo_command_ce, &options) == FAILURE) {
@@ -441,7 +473,12 @@ static PHP_METHOD(Manager, executeReadWriteCommand)
 
 	intern = Z_MANAGER_OBJ_P(getThis());
 
-	if (!php_phongo_manager_select_server(true, NULL, intern->client, &server_id TSRMLS_CC)) {
+	if (!phongo_parse_session(options, intern->client, NULL, &zsession TSRMLS_CC)) {
+		/* Exception should already have been thrown */
+		return;
+	}
+
+	if (!php_phongo_manager_select_server(true, NULL, zsession, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		return;
 	}
@@ -461,6 +498,7 @@ static PHP_METHOD(Manager, executeQuery)
 	bool                free_options    = false;
 	zval*               zreadPreference = NULL;
 	uint32_t            server_id       = 0;
+	zval*               zsession        = NULL;
 	DECLARE_RETURN_VALUE_USED
 
 	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sO|z!", &namespace, &namespace_len, &query, php_phongo_query_ce, &options) == FAILURE) {
@@ -471,12 +509,17 @@ static PHP_METHOD(Manager, executeQuery)
 
 	options = php_phongo_prep_legacy_option(options, "readPreference", &free_options TSRMLS_CC);
 
+	if (!phongo_parse_session(options, intern->client, NULL, &zsession TSRMLS_CC)) {
+		/* Exception should already have been thrown */
+		goto cleanup;
+	}
+
 	if (!phongo_parse_read_preference(options, &zreadPreference TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		goto cleanup;
 	}
 
-	if (!php_phongo_manager_select_server(false, zreadPreference, intern->client, &server_id TSRMLS_CC)) {
+	if (!php_phongo_manager_select_server(false, zreadPreference, zsession, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		goto cleanup;
 	}
@@ -501,6 +544,7 @@ static PHP_METHOD(Manager, executeBulkWrite)
 	zval*                   options      = NULL;
 	bool                    free_options = false;
 	uint32_t                server_id    = 0;
+	zval*                   zsession     = NULL;
 	DECLARE_RETURN_VALUE_USED
 
 	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sO|z!", &namespace, &namespace_len, &zbulk, php_phongo_bulkwrite_ce, &options) == FAILURE) {
@@ -512,7 +556,12 @@ static PHP_METHOD(Manager, executeBulkWrite)
 
 	options = php_phongo_prep_legacy_option(options, "writeConcern", &free_options TSRMLS_CC);
 
-	if (!php_phongo_manager_select_server(true, NULL, intern->client, &server_id TSRMLS_CC)) {
+	if (!phongo_parse_session(options, intern->client, NULL, &zsession TSRMLS_CC)) {
+		/* Exception should already have been thrown */
+		return;
+	}
+
+	if (!php_phongo_manager_select_server(true, NULL, zsession, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		goto cleanup;
 	}
@@ -628,7 +677,7 @@ static PHP_METHOD(Manager, selectServer)
 		return;
 	}
 
-	if (!php_phongo_manager_select_server(false, zreadPreference, intern->client, &server_id TSRMLS_CC)) {
+	if (!php_phongo_manager_select_server(false, zreadPreference, NULL, intern->client, &server_id TSRMLS_CC)) {
 		/* Exception should already have been thrown */
 		return;
 	}

src/MongoDB/Session.c

@@ -38,18 +38,6 @@ zend_class_entry* php_phongo_session_ce;
 		return;                                                         \
 	}
 
-static bool php_phongo_topology_is_sharded_cluster(mongoc_client_t* client)
-{
-	mongoc_server_description_t* sd;
-	bool ret;
-
-	sd = mongoc_client_select_server(client, true, NULL, NULL);
-	ret = (sd && !strcmp(mongoc_server_description_type(sd), php_phongo_server_description_type_map[PHONGO_SERVER_MONGOS].name));
-	mongoc_server_description_destroy(sd);
-
-	return ret;
-}
-
 static bool php_phongo_session_get_timestamp_parts(zval* obj, uint32_t* timestamp, uint32_t* increment TSRMLS_DC)
 {
 	bool retval = false;
@@ -246,6 +234,30 @@ static PHP_METHOD(Session, getOperationTime)
 	php_phongo_new_timestamp_from_increment_and_timestamp(return_value, increment, timestamp TSRMLS_CC);
 } /* }}} */
 
+/* {{{ proto MongoDB\Driver\Server|null MongoDB\Driver\Session::getServer()
+   Returns the server this session is pinned to */
+static PHP_METHOD(Session, getServer)
+{
+	php_phongo_session_t* intern;
+	uint32_t              server_id = 0;
+
+	intern = Z_SESSION_OBJ_P(getThis());
+	SESSION_CHECK_LIVELINESS(intern, "getServer")
+
+	if (zend_parse_parameters_none() == FAILURE) {
+		return;
+	}
+
+	server_id = mongoc_client_session_get_server_id(intern->client_session);
+
+	/* For sessions without a pinned server, 0 is returned. */
+	if (!server_id) {
+	    RETURN_NULL();
+	}
+
+	phongo_server_init(return_value, mongoc_client_session_get_client(intern->client_session), server_id TSRMLS_CC);
+} /* }}} */
+
 /* Creates a opts structure from an array optionally containing an RP, RC,
  * WC object, and/or maxCommitTimeMS int. Returns NULL if no options were found,
  * or there was an invalid option. If there was an invalid option or structure,
@@ -366,11 +378,6 @@ static PHP_METHOD(Session, startTransaction)
 		return;
 	}
 
-	if (php_phongo_topology_is_sharded_cluster(mongoc_client_session_get_client(intern->client_session))) {
-		phongo_throw_exception(PHONGO_ERROR_RUNTIME TSRMLS_CC, "PHP MongoDB driver %s does not support running multi-document transactions on sharded clusters", PHP_MONGODB_VERSION);
-		return;
-	}
-
 	if (!mongoc_client_session_start_transaction(intern->client_session, txn_options, &error)) {
 		phongo_throw_exception_from_bson_error_t(&error TSRMLS_CC);
 	}
@@ -478,6 +485,7 @@ static zend_function_entry php_phongo_session_me[] = {
 	PHP_ME(Session, getClusterTime, ai_Session_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Session, getLogicalSessionId, ai_Session_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Session, getOperationTime, ai_Session_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
+	PHP_ME(Session, getServer, ai_Session_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Session, isInTransaction, ai_Session_void, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	PHP_ME(Session, startTransaction, ai_Session_startTransaction, ZEND_ACC_PUBLIC | ZEND_ACC_FINAL)
 	ZEND_NAMED_ME(__construct, PHP_FN(MongoDB_disabled___construct), ai_Session_void, ZEND_ACC_PRIVATE | ZEND_ACC_FINAL)
@@ -618,6 +626,30 @@ static HashTable* php_phongo_session_get_debug_info(zval* object, int* is_temp T
 		ADD_ASSOC_NULL_EX(&retval, "operationTime");
 	}
 
+	if (intern->client_session) {
+		uint32_t server_id = mongoc_client_session_get_server_id(intern->client_session);
+
+		if (server_id) {
+
+#if PHP_VERSION_ID >= 70000
+			zval server;
+
+			phongo_server_init(&server, mongoc_client_session_get_client(intern->client_session), server_id TSRMLS_CC);
+			ADD_ASSOC_ZVAL_EX(&retval, "server", &server);
+#else
+			zval* server = NULL;
+
+			MAKE_STD_ZVAL(server);
+			phongo_server_init(server, mongoc_client_session_get_client(intern->client_session), server_id TSRMLS_CC);
+			ADD_ASSOC_ZVAL_EX(&retval, "server", server);
+#endif
+		} else {
+			ADD_ASSOC_NULL_EX(&retval, "server");
+		}
+	} else {
+		ADD_ASSOC_NULL_EX(&retval, "server");
+	}
+
 	return Z_ARRVAL(retval);
 } /* }}} */
 /* }}} */

src/libmongoc

@@ -3,7 +3,7 @@ Causal consistency: new session has no operation time
 --SKIPIF--
 <?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
 <?php skip_if_not_libmongoc_crypto(); ?>
-<?php skip_if_not_replica_set(); ?>
+<?php skip_if_not_replica_set_or_mongos_with_replica_set(); ?>
 <?php skip_if_server_version('<', '3.6'); ?>
 --FILE--
 <?php

tests/causal-consistency/causal-consistency-001.phpt

@@ -3,7 +3,7 @@ Causal consistency: first read in session does not include afterClusterTime
 --SKIPIF--
 <?php require __DIR__ . "/../utils/basic-skipif.inc"; ?>
 <?php skip_if_not_libmongoc_crypto(); ?>
-<?php skip_if_not_replica_set(); ?>
+<?php skip_if_not_replica_set_or_mongos_with_replica_set(); ?>
 <?php skip_if_server_version('<', '3.6'); ?>
 --FILE--
 <?php