feat: Add support for batch session creation in Cloud Spanner

Spanner/src/Connection/ConnectionInterface.php

@@ -130,6 +130,11 @@ public function createSession(array $args);
      */
     public function createSessionAsync(array $args);
 
+    /**
+     * @param array $args
+     */
+    public function batchCreateSessions(array $args);
+
     /**
      * @param array $args
      */

Spanner/src/Connection/Grpc.php

@@ -492,6 +492,24 @@ public function createSessionAsync(array $args)
         );
     }
 
+    /**
+     * @param array $args
+     */
+    public function batchCreateSessions(array $args)
+    {
+        $args['sessionTemplate'] = $this->serializer->decodeMessage(
+            new Session,
+            $this->pluck('sessionTemplate', $args)
+        );
+
+        $database = $this->pluck('database', $args);
+        return $this->send([$this->spannerClient, 'batchCreateSessions'], [
+            $database,
+            $this->pluck('sessionCount', $args),
+            $this->addResourcePrefixHeader($args, $database)
+        ]);
+    }
+
     /**
      * @param array $args
      */

Spanner/src/Session/CacheSessionPool.php

@@ -247,7 +247,7 @@ public function acquire($context = SessionPoolInterface::CONTEXT_READ)
 
         // Create a session if needed.
         if ($toCreate) {
-            $createdSessions = $this->createSessions(count($toCreate));
+            $createdSessions = $this->createSessions(count($toCreate))[0];
             $hasCreatedSessions = count($createdSessions) > 0;
 
             $session = $this->config['lock']->synchronize(function () use (
@@ -429,13 +429,8 @@ public function warmup()
             return 0;
         }
 
-        $createdSessions = [];
         $exception = null;
-
-        try {
-            $createdSessions = $this->createSessions(count($toCreate));
-        } catch (\Exception $exception) {
-        }
+        list ($createdSessions, $exception) = $this->createSessions(count($toCreate));
 
         $this->config['lock']->synchronize(function () use ($toCreate, $createdSessions) {
             $item = $this->cacheItemPool->getItem($this->cacheKey);
@@ -639,38 +634,40 @@ private function getSession(array &$data)
      * Creates sessions up to the count provided.
      *
      * @param int $count
-     * @return array
+     * @return [ array[] $sessions, \Exception $ex = null ]
      */
     private function createSessions($count)
     {
-        $args = [
-            'database' => $this->database->name(),
-            'session' => [
-                'labels' => isset($this->config['labels']) ? $this->config['labels'] : []
-            ]
-        ];
-
-        $promises = [];
-
-        for ($i = 0; $i < $count; $i++) {
-            $promises[] = $this->database->connection()->createSessionAsync($args);
-        }
-
-        $results = Promise\settle($promises)->wait();
-
         $sessions = [];
+        $created = 0;
+        $exception = null;
+
+        // Loop over RPC in case it returns less than the desired number of sessions.
+        // @see https://github.com/googleapis/google-cloud-php/pull/2342#discussion_r327925546
+        while ($count > $created) {
+            try {
+                $res = $this->database->connection()->batchCreateSessions([
+                    'database' => $this->database->name(),
+                    'sessionTemplate' => [
+                        'labels' => isset($this->config['labels']) ? $this->config['labels'] : []
+                    ],
+                    'sessionCount' => $count - $created
+                ]);
+            } catch (\Exception $exception) {
+                break;
+            }
 
-        foreach ($results as $result) {
-            if ($result['state'] === 'fulfilled') {
-                $name = $result['value']->getName();
+            foreach ($res['session'] as $result) {
                 $sessions[] = [
-                    'name' => $name,
+                    'name' => $result['name'],
                     'expiration' => $this->time() + SessionPoolInterface::SESSION_EXPIRATION_SECONDS
                 ];
+
+                $created++;
             }
         }
 
-        return $sessions;
+        return [$sessions, $exception];
     }
 
     /**

Spanner/tests/System/SessionTest.php

@@ -0,0 +1,102 @@
+<?php
+/**
+ * Copyright 2019 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace Google\Cloud\Spanner\Tests\System;
+
+use Google\Auth\Cache\MemoryCacheItemPool;
+use Google\Cloud\Spanner\Session\CacheSessionPool;
+use Google\Cloud\Spanner\Session\Session;
+use Psr\Cache\CacheItemPoolInterface;
+
+/**
+ * @group spanner
+ * @group spanner-session
+ */
+class SessionTest extends SpannerTestCase
+{
+    public function testCacheSessionPool()
+    {
+        $identity = self::$database->identity();
+        $cacheKey = sprintf(
+            CacheSessionPool::CACHE_KEY_TEMPLATE,
+            $identity['projectId'],
+            $identity['instance'],
+            $identity['database']
+        );
+
+        $cache = new MemoryCacheItemPool;
+        $pool = new CacheSessionPool($cache, [
+            'maxSessions' => 10,
+            'minSessions' => 5,
+            'shouldWaitForSession' => false
+        ]);
+        $pool->setDatabase(self::$database);
+
+        $this->assertNull($cache->getItem($cacheKey)->get());
+
+        $pool->warmup();
+
+        $this->assertPoolCounts($cache, $cacheKey, 5, 0, 0);
+
+        $session = $pool->acquire();
+        $this->assertInstanceOf(Session::class, $session);
+        $this->assertTrue($session->exists());
+        $this->assertPoolCounts($cache, $cacheKey, 4, 1, 0);
+        $this->assertEquals($session->name(), current($cache->getItem($cacheKey)->get()['inUse'])['name']);
+
+        $pool->release($session);
+
+        $inUse = [];
+        for ($i = 0; $i < 10; $i++) {
+            $inUse[] = $pool->acquire();
+        }
+
+        $this->assertPoolCounts($cache, $cacheKey, 0, 10, 0);
+
+        $exception = null;
+        try {
+            $pool->acquire();
+        } catch (\RuntimeException $exception) {
+            // no-op
+        }
+        $this->assertInstanceOf(
+            \RuntimeException::class,
+            $exception,
+            'Should catch a RuntimeException when pool is exhausted.'
+        );
+
+        foreach ($inUse as $i) {
+            $pool->release($i);
+        }
+        sleep(1);
+
+        $this->assertPoolCounts($cache, $cacheKey, 10, 0, 0);
+
+        $pool->clear();
+        sleep(1);
+        $this->assertNull($cache->getItem($cacheKey)->get());
+        $this->assertFalse($inUse[0]->exists());
+    }
+
+    private function assertPoolCounts(CacheItemPoolInterface $cache, $key, $queue, $inUse, $toCreate)
+    {
+        $item = $cache->getItem($key)->get();
+        $this->assertCount($queue, $item['queue'], 'Sessions In Queue');
+        $this->assertCount($inUse, $item['inUse'], 'Sessions In Use');
+        $this->assertCount($toCreate, $item['toCreate'], 'Sessions To Create');
+    }
+}

Spanner/tests/Unit/Batch/BatchClientTest.php

@@ -90,9 +90,6 @@ public function testSnapshot()
         $this->assertInstanceOf(BatchSnapshot::class, $snapshot);
     }
 
-    /**
-     * @group foo
-     */
     public function testSnapshotFromString()
     {
         $time = time();

Spanner/tests/Unit/Connection/GrpcTest.php

@@ -344,6 +344,26 @@ public function testCreateSessionAsync()
         $this->assertInstanceOf(PromiseInterface::class, $promise);
     }
 
+    public function testBatchCreateSessions()
+    {
+        $count = 10;
+        $template = [
+            'labels' => [
+                'foo' => 'bar'
+            ]
+        ];
+
+        $this->assertCallCorrect('batchCreateSessions', [
+            'database' => self::DATABASE,
+            'sessionCount' => $count,
+            'sessionTemplate' => $template
+        ], $this->expectResourceHeader(self::DATABASE, [
+            self::DATABASE, $count, [
+                'sessionTemplate' => $this->serializer->decodeMessage(new Session, $template)
+            ]
+        ]));
+    }
+
     public function testGetSession()
     {
         $this->assertCallCorrect('getSession', [

Spanner/tests/Unit/Session/CacheSessionPoolTest.php

@@ -25,7 +25,6 @@
 use Google\Cloud\Spanner\Session\Session;
 use Google\Cloud\Core\Testing\GrpcTestTrait;
 use GuzzleHttp\Promise\PromiseInterface;
-use GuzzleHttp\Promise\RejectedPromise;
 use Psr\Cache\CacheItemPoolInterface;
 use Prophecy\Argument;
 use Prophecy\Argument\ArgumentsWildcard;
@@ -321,7 +320,7 @@ public function testWarmup()
             $this->getCacheItemPool(),
             ['minSessions' => $expectedCreationCount]
         );
-        $pool->setDatabase($this->getDatabase());
+        $pool->setDatabase($this->getDatabase(false, false, 5));
         $response = $pool->warmup();
 
         $this->assertEquals($expectedCreationCount, $response);
@@ -669,7 +668,7 @@ public function acquireDataProvider()
         ];
     }
 
-    private function getDatabase($shouldCreateFails = false, $willDeleteSessions = false)
+    private function getDatabase($shouldCreateFails = false, $willDeleteSessions = false, $expectedCreateCalls = null)
     {
         $database = $this->prophesize(Database::class);
         $session = $this->prophesize(Session::class);
@@ -708,20 +707,33 @@ private function getDatabase($shouldCreateFails = false, $willDeleteSessions = f
             ->willReturn(self::DATABASE_NAME);
 
         $createdSession = $this->prophesize(\Google\Cloud\Spanner\V1\Session::class);
-        $connection->createSessionAsync(Argument::any())
-            ->will(function ($args, $mock, $method) use ($createdSession, $shouldCreateFails) {
-                if ($shouldCreateFails) {
-                    return new RejectedPromise("error");
-                }
-
-                $methodCalls = $mock->findProphecyMethodCalls(
-                    $method->getMethodName(),
-                    new ArgumentsWildcard($args)
-                );
-
-                $createdSession->getName()->willReturn('session' . count($methodCalls));
-                return $createdSession->reveal();
-            });
+        $createRes = function ($args, $mock, $method) use ($createdSession, $shouldCreateFails) {
+            if ($shouldCreateFails) {
+                throw new \Exception("error");
+            }
+
+            $methodCalls = $mock->findProphecyMethodCalls(
+                $method->getMethodName(),
+                new ArgumentsWildcard($args)
+            );
+
+            return [
+                'session' => [
+                    [
+                        'name' => 'session' . count($methodCalls)
+                    ]
+                ]
+            ];
+        };
+
+        if ($expectedCreateCalls) {
+            $connection->batchCreateSessions(Argument::any())
+                ->shouldBeCalledTimes($expectedCreateCalls)
+                ->will($createRes);
+        } else {
+            $connection->batchCreateSessions(Argument::any())
+                ->will($createRes);
+        }
 
         return $database->reveal();
     }

Spanner/tests/Unit/ValueMapperTest.php

@@ -643,7 +643,6 @@ public function testFormatParamsForExecuteSqlStdClassMissingDefinition()
 
     /**
      * @dataProvider simpleTypeValues
-     * @group foo
      */
     public function testEncodeValuesAsSimpleType($value, $expected = null)
     {