API-553 Compact serialization and Generic Records

benchmark/map_latency_benchmark/index.js

@@ -0,0 +1,127 @@
+/*
+ * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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.
+ */
+'use strict';
+
+const { Client } = require('../../lib');
+const fs = require('fs');
+
+const CLUSTER_MEMBERS = ['10.212.1.117', '10.212.1.118'];
+// const CLUSTER_MEMBERS = ['127.0.0.1'];
+const KEY = 1;
+const VALUE = 2;
+const BENCHMARK_DURATION_SECONDS = 10;
+const MAP_NAME = 'test';
+const DEFAULT_NUMBER_TYPE = 'integer';
+
+let benchmarkEnded = false;
+let warmupEnded = false;
+
+async function createClient() {
+    const client = await Client.newHazelcastClient({
+        serialization: {
+            defaultNumberType: DEFAULT_NUMBER_TYPE
+        },
+        network: {
+            clusterMembers: CLUSTER_MEMBERS
+        }
+    });
+    return client;
+}
+
+async function runPut(map) {
+    const start = process.hrtime.bigint();
+    await map.put(KEY, VALUE);
+    const end = process.hrtime.bigint();
+    return end - start;
+}
+
+async function runGet(map) {
+    const start = process.hrtime.bigint();
+    await map.get(KEY);
+    const end = process.hrtime.bigint();
+    return end - start;
+}
+
+async function runBenchMark(timeArray, client, runBenchmarkFn) {
+    const map = await client.getMap(MAP_NAME);
+
+    console.log('Warmup started for ', runBenchmarkFn.name);
+    while (!warmupEnded) {
+        await runBenchmarkFn(map);
+    }
+
+    console.log('Benchmark started for ', runBenchmarkFn.name);
+    while (!benchmarkEnded) {
+        const timeElapsed = await runBenchmarkFn(map);
+        timeArray.push(timeElapsed);
+    }
+}
+
+function setBenchmarkAsStartedAndSetTimer() {
+    benchmarkEnded = false;
+    warmupEnded = false;
+    const warmupDuration = BENCHMARK_DURATION_SECONDS * 1000 / 5;
+    const benchmarkDuration = warmupDuration + BENCHMARK_DURATION_SECONDS * 1000;
+
+    setTimeout(() => {
+        warmupEnded = true;
+    }, warmupDuration);
+    setTimeout(() => {
+        benchmarkEnded = true;
+    }, benchmarkDuration);
+}
+
+function convertToNumbersAndCalculateAverage(timeArray) {
+    const total = timeArray.reduce((acc, current) => {
+        return acc + current;
+    }, 0);
+    const average = Math.trunc(total / timeArray.length);
+    return average;
+}
+
+function writeToFile(filename, average, timeArray) {
+    fs.writeFileSync(filename, JSON.stringify({
+        average,
+        timeArray
+    }));
+}
+
+async function runBenchMarkAndLogToFile(client, runFn, runBenchmarkFn, filename) {
+    const timeArray = [];
+
+    setBenchmarkAsStartedAndSetTimer();
+
+    await runFn(timeArray, client, runBenchmarkFn);
+
+    const timeArrayNumbers = timeArray.map(x => Number(x));
+    const average = convertToNumbersAndCalculateAverage(timeArrayNumbers);
+
+    writeToFile(filename, average, timeArrayNumbers);
+}
+
+async function main() {
+    const client = await createClient();
+
+    await runBenchMarkAndLogToFile(client, runBenchMark, runPut, 'put.json');
+    await runBenchMarkAndLogToFile(client, runBenchMark, runGet, 'get.json');
+
+    await client.shutdown();
+}
+
+main().catch(e => {
+    console.error(e);
+    process.exit(1);
+});

scripts/code-sample-checker.js

@@ -48,7 +48,7 @@ exports.main = async (cluster) => {
     console.log(`Will run ${JSON.stringify(files)}`);
 
     const numberOfFiles = files.length;
-    let counter = 0;
+    let counter = 1;
     for (const file of files) {
         // start and terminate for each code sample to avoid map name clashes
         const member = await RC.startMember(cluster.id);

scripts/test-runner.js

@@ -37,7 +37,7 @@ let runTests = true;
 let CLASSPATH = `hazelcast-remote-controller-${HAZELCAST_RC_VERSION}.jar${PATH_SEPARATOR}`
               + `hazelcast-${HAZELCAST_TEST_VERSION}-tests.jar${PATH_SEPARATOR}`
               + `hazelcast-sql-${HAZELCAST_VERSION}.jar${PATH_SEPARATOR}`
-              + 'test/javaclasses';
+              + 'test/integration/resources/compact-test-resources-1.0.jar';
 
 if (HAZELCAST_ENTERPRISE_KEY) {
     CLASSPATH = `hazelcast-enterprise-${HAZELCAST_ENTERPRISE_VERSION}.jar${PATH_SEPARATOR}`

src/HazelcastClient.ts

@@ -66,6 +66,7 @@ import {ClientMessage} from './protocol/ClientMessage';
 import {Connection} from './network/Connection';
 import {ConnectionRegistryImpl} from './network/ConnectionRegistry';
 import {SqlService, SqlServiceImpl} from './sql/SqlService';
+import {SchemaService} from './serialization/compact/SchemaService';
 
 /**
  * Hazelcast client instance. When you want to use Hazelcast's distributed
@@ -129,6 +130,8 @@ export class HazelcastClient {
     private readonly sqlService: SqlService;
     /** @internal */
     private shutdownPromise: Promise<void> | undefined;
+    /** @internal */
+    private readonly schemaService: SchemaService;
 
     /** @internal */
     constructor(config?: ClientConfigImpl, failoverConfig?: ClientFailoverConfigImpl) {
@@ -140,26 +143,28 @@ export class HazelcastClient {
         this.loadBalancer = this.initLoadBalancer();
         this.failoverConfig = failoverConfig;
         this.errorFactory = new ClientErrorFactory();
-        this.serializationService = new SerializationServiceV1(this.config.serialization);
-        this.instanceName = this.config.instanceName || 'hz.client_' + this.id;
         this.loggingService = new LoggingService(this.config.customLogger,
             this.config.properties['hazelcast.logging.level'] as string);
+        const logger = this.loggingService.getLogger();
+        this.schemaService = new SchemaService(() => this.invocationService, logger);
+        this.serializationService = new SerializationServiceV1(this.config.serialization, this.schemaService);
+        this.instanceName = this.config.instanceName || 'hz.client_' + this.id;
         this.nearCacheManager = new NearCacheManager(
             this.config,
             this.serializationService
         );
         this.partitionService = new PartitionServiceImpl(
-            this.loggingService.getLogger(),
+            logger,
             this.serializationService
         );
         this.lifecycleService = new LifecycleServiceImpl(
             this.config.lifecycleListeners,
-            this.loggingService.getLogger()
+            logger
         );
         this.clusterFailoverService = this.initClusterFailoverService();
         this.clusterService = new ClusterService(
             this.config,
-            this.loggingService.getLogger(),
+            logger,
             this.clusterFailoverService
         );
         this.connectionRegistry = new ConnectionRegistryImpl(
@@ -171,17 +176,19 @@ export class HazelcastClient {
         );
         this.invocationService = new InvocationService(
             this.config,
-            this.loggingService.getLogger(),
+            logger,
             this.partitionService as PartitionServiceImpl,
             this.errorFactory,
             this.lifecycleService,
-            this.connectionRegistry
+            this.connectionRegistry,
+            this.schemaService,
+            this.serializationService
         );
         this.connectionManager = new ConnectionManager(
             this,
             this.instanceName,
             this.config,
-            this.loggingService.getLogger(),
+            logger,
             this.partitionService,
             this.serializationService,
             this.lifecycleService,
@@ -192,15 +199,15 @@ export class HazelcastClient {
             this.connectionRegistry
         );
         this.listenerService = new ListenerService(
-            this.loggingService.getLogger(),
+            logger,
             this.config.network.smartRouting,
             this.connectionManager,
             this.invocationService
         );
         this.lockReferenceIdGenerator = new LockReferenceIdGenerator();
         this.proxyManager = new ProxyManager(
             this.config,
-            this.loggingService.getLogger(),
+            logger,
             this.invocationService,
             this.listenerService,
             this.partitionService,
@@ -209,7 +216,8 @@ export class HazelcastClient {
             () => this.getRepairingTask(),
             this.clusterService,
             this.lockReferenceIdGenerator,
-            this.connectionRegistry
+            this.connectionRegistry,
+            this.schemaService
         );
         this.statistics = new Statistics(
             this.loggingService.getLogger(),
@@ -220,14 +228,14 @@ export class HazelcastClient {
             this.connectionManager
         );
         this.clusterViewListenerService = new ClusterViewListenerService(
-            this.loggingService.getLogger(),
+            logger,
             this.connectionManager,
             this.partitionService as PartitionServiceImpl,
             this.clusterService,
             this.invocationService
         );
         this.cpSubsystem = new CPSubsystemImpl(
-            this.loggingService.getLogger(),
+            logger,
             this.instanceName,
             this.invocationService,
             this.serializationService
@@ -304,12 +312,11 @@ export class HazelcastClient {
         const clientMessage = ClientGetDistributedObjectsCodec.encodeRequest();
         let localDistributedObjects: Set<string>;
         let responseMessage: ClientMessage;
-        return this.invocationService.invokeOnRandomTarget(clientMessage)
+        return this.invocationService.invokeOnRandomTarget(clientMessage, x => x)
             .then((resp) => {
                 responseMessage = resp;
                 return this.proxyManager.getDistributedObjects();
-            })
-            .then((distributedObjects) => {
+            }).then((distributedObjects) => {
                 localDistributedObjects = new Set<string>();
                 distributedObjects.forEach((obj) => {
                     localDistributedObjects.add(obj.getServiceName() + NAMESPACE_SEPARATOR + obj.getName());
@@ -562,7 +569,9 @@ export class HazelcastClient {
 
     /** @internal */
     sendStateToCluster(): Promise<void> {
-        return this.proxyManager.createDistributedObjectsOnCluster();
+        return this.schemaService.sendAllSchemas().then(() => {
+            return this.proxyManager.createDistributedObjectsOnCluster();
+        });
     }
 
     /** @internal */

src/PartitionService.ts

@@ -47,6 +47,7 @@ export interface PartitionService {
      * @param key
      * @returns the partition id.
      * @throws {@link ClientOfflineError} if the partition table has not arrived yet.
+     * @throws {@link HazelcastSerializationError} if key cannot be serialized
      */
     getPartitionId(key: any): number;
 
@@ -106,7 +107,7 @@ export class PartitionServiceImpl implements PartitionService {
         if (typeof key === 'object' && 'getPartitionHash' in key) {
             partitionHash = key.getPartitionHash();
         } else {
-            partitionHash = this.serializationService.toData(key).getPartitionHash();
+            partitionHash = this.serializationService.toData(key, undefined, false).getPartitionHash();
         }
         return Math.abs(partitionHash) % this.partitionCount;
     }

src/codec/ClientFetchSchemaCodec.ts

@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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.
+ */
+
+/* eslint-disable max-len */
+import {BitsUtil} from '../util/BitsUtil';
+import {FixSizedTypesCodec} from './builtin/FixSizedTypesCodec';
+import {ClientMessage, Frame, PARTITION_ID_OFFSET} from '../protocol/ClientMessage';
+import * as Long from 'long';
+import {Schema} from '../serialization/compact/Schema';
+import {SchemaCodec} from './custom/SchemaCodec';
+import {CodecUtil} from './builtin/CodecUtil';
+
+// hex: 0x001400
+const REQUEST_MESSAGE_TYPE = 5120;
+// hex: 0x001401
+// RESPONSE_MESSAGE_TYPE = 5121
+
+const REQUEST_SCHEMA_ID_OFFSET = PARTITION_ID_OFFSET + BitsUtil.INT_SIZE_IN_BYTES;
+const REQUEST_INITIAL_FRAME_SIZE = REQUEST_SCHEMA_ID_OFFSET + BitsUtil.LONG_SIZE_IN_BYTES;
+
+/** @internal */
+export class ClientFetchSchemaCodec {
+    static encodeRequest(schemaId: Long): ClientMessage {
+        const clientMessage = ClientMessage.createForEncode();
+        clientMessage.setRetryable(true);
+
+        const initialFrame = Frame.createInitialFrame(REQUEST_INITIAL_FRAME_SIZE);
+        FixSizedTypesCodec.encodeLong(initialFrame.content, REQUEST_SCHEMA_ID_OFFSET, schemaId);
+        clientMessage.addFrame(initialFrame);
+        clientMessage.setMessageType(REQUEST_MESSAGE_TYPE);
+        clientMessage.setPartitionId(-1);
+
+        return clientMessage;
+    }
+
+    static decodeResponse(clientMessage: ClientMessage): Schema {
+        // empty initial frame
+        clientMessage.nextFrame();
+
+        return CodecUtil.decodeNullable(clientMessage, SchemaCodec.decode);
+    }
+}