- cache hydrated objects per-remote peer in the ObjectDiscoveryAgent

- prevent re-hashing stuff in other places

Author: sbazerque

src/data/model/hashing/Hashing.ts

@@ -15,14 +15,21 @@ class Hashing {
             seed = '';
         }
 
-        let firstPass  = Hashing.sha.sha256base64('0a' + text + seed);
-        let secondPass = Hashing.rmd.rmd160base64(text + firstPass); 
+        //const t = Date.now();
+
+        const firstPass  = Hashing.sha.sha256base64('0a' + text + seed);
+        const secondPass = Hashing.rmd.rmd160base64(text + firstPass); 
+
+        //console.trace();
+        //console.log(' *** hashing took ', Date.now() - t, ' for result ', secondPass);
 
         return secondPass;
     }
 
     static forValue(value: any, seed?: string) : Hash{
-        let text = Serialization.default(value);
+        //const t = Date.now()
+        const text = Serialization.default(value);
+        //console.log(' *** serialization took ', Date.now() - t);
 
         return Hashing.forString(text, seed);
     }

src/data/model/immutable/HashedObject.ts

@@ -205,6 +205,9 @@ abstract class HashedObject {
 
     hash(seed?: string): Hash {
 
+        //console.log('about to hash a ' + this.getClassName())
+        //console.trace();
+
         let hash = this.customHash(seed);
 
         if (hash === undefined) {

src/mesh/agents/discovery/ObjectDiscoveryAgent.ts

@@ -7,6 +7,7 @@ import { Endpoint, LinkupMessage, NetworkAgent, NetworkEventType } from '../netw
 import { ObjectBroadcastAgent, ObjectBroadcastRequest, ObjectBroadcastReply } from './ObjectBroadcastAgent';
 
 import { AsyncStream, BufferedAsyncStream, BufferingAsyncStreamSource, AsyncStreamSource, FilteredAsyncStreamSource } from 'util/streams';
+import { LRUCache } from 'util/caching';
 
 type Params = {
     broadcastedSuffixBits : number,
@@ -28,6 +29,10 @@ class ObjectDiscoveryAgent implements Agent {
 
     static newestReplyFirst = (a: ObjectDiscoveryReply, b: ObjectDiscoveryReply) => (b.timestamp - a.timestamp);
 
+    private static makeEndpointObjectPair(hash: Hash, source: Endpoint, destination: Endpoint) {
+        return hash + '_' + source.replaceAll('_', '__') + '_' + destination.replaceAll('_', '__');
+    }
+
     pod?: AgentPod;
 
 
@@ -39,6 +44,7 @@ class ObjectDiscoveryAgent implements Agent {
 
     streamSource: BufferingAsyncStreamSource<ObjectDiscoveryReply>;
 
+    discoveredEndpointObjectPairs: LRUCache<string, HashedObject>;
 
     wasShutdown = false;
 
@@ -60,6 +66,8 @@ class ObjectDiscoveryAgent implements Agent {
         this.lastQueryingTimePerServer = new Map();
 
         this.streamSource = new BufferingAsyncStreamSource(this.params.maxStoredReplies);
+
+        this.discoveredEndpointObjectPairs = new LRUCache(256);
     }
 
     getAgentId(): string {
@@ -169,52 +177,81 @@ class ObjectDiscoveryAgent implements Agent {
 
             if (msg.agentId === this.getAgentId()) {
 
-                const reply = msg.content as ObjectBroadcastReply;
-
 
 
-                let replyHash: Hash|undefined = undefined;
-                let object: HashedObject | undefined = undefined;
-                let error: string | undefined = undefined;
-                try {
-                    object = HashedObject.fromLiteralContext(reply.literalContext);
-                    replyHash = object.hash();
-                } catch (e: any) {
+                this.processReply(msg);
+
+            }
 
-                    // Since anybody may reply with _anything_, only replies that at least match the hash
-                    // suffix being queried will be reported back to the caller.
+        }
+    }
 
-                    ObjectDiscoveryAgent.log.warning('Error deliteralizing object discovery reply:' + e);
-                    object = undefined;
-                    replyHash = undefined;
-                    const literal = reply.literalContext.literals[reply.literalContext.rootHashes[0]];
+    private processReply(msg: LinkupMessage) {
 
-                    if (literal !== undefined && LiteralUtils.validateHash(literal)) {
-                        if (literal.hash === reply.literalContext.rootHashes[0]) {
-                            replyHash = literal.hash;
-                        }
-                    }
+        const reply = msg.content as ObjectBroadcastReply;
 
-                    error = String(e);
-                }
+        let replyHash: Hash|undefined = undefined;
+        let object: HashedObject | undefined = undefined;
+        let error: string | undefined = undefined;
 
-                if (replyHash === reply.literalContext.rootHashes[0] && 
-                    this.hexHashSuffix === Hashing.toHex(replyHash).slice(-this.hexHashSuffix.length) &&
-                    this.localEndpoints.has(msg.destination)) {
+        
 
-                    ObjectDiscoveryAgent.log.trace(() => 'Received object with hash ' + replyHash + ' from ' + msg.source + ' at ' + msg.destination);
+        try {
+            const declaredHash = reply.literalContext.rootHashes.length === 1? 
+                                        reply.literalContext.rootHashes[0]
+                                    :
+                                        '';
 
-                    let item = {source: msg.source, destination: msg.destination, hash: replyHash, object: object, error: error, timestamp: Date.now()}; 
-                    
+            if (this.hexHashSuffix === Hashing.toHex(declaredHash).slice(-this.hexHashSuffix.length)) {
+               
+                const cacheKey = ObjectDiscoveryAgent.makeEndpointObjectPair(declaredHash, msg.source, msg.destination);
+                object = this.discoveredEndpointObjectPairs.get(cacheKey);
 
-                    this.streamSource.ingest(item);
-                    
+                if (object === undefined) {
+                    object = HashedObject.fromLiteralContext(reply.literalContext);
+                    replyHash = object.hash();
+
+                    if (replyHash === declaredHash) {
+                        this.discoveredEndpointObjectPairs.set(cacheKey, object);
+                    }
                 } else {
-                    ObjectDiscoveryAgent.log.debug('Error validating object discovery reply');
+                    replyHash = declaredHash;
                 }
+            }
+
+            
+        } catch (e: any) {
+
+            // Since anybody may reply with _anything_, only replies that at least match the hash
+            // suffix being queried will be reported back to the caller.
+
+            ObjectDiscoveryAgent.log.warning('Error deliteralizing object discovery reply:' + e);
+            object = undefined;
+            replyHash = undefined;
+            const literal = reply.literalContext.literals[reply.literalContext.rootHashes[0]];
 
+            if (literal !== undefined && LiteralUtils.validateHash(literal)) {
+                if (literal.hash === reply.literalContext.rootHashes[0]) {
+                    replyHash = literal.hash;
+                }
             }
 
+            error = String(e);
+        }
+
+        if (replyHash === reply.literalContext.rootHashes[0] && 
+            this.hexHashSuffix === Hashing.toHex(replyHash).slice(-this.hexHashSuffix.length) &&
+            this.localEndpoints.has(msg.destination)) {
+
+            ObjectDiscoveryAgent.log.trace(() => 'Received object with hash ' + replyHash + ' from ' + msg.source + ' at ' + msg.destination);
+
+            let item = {source: msg.source, destination: msg.destination, hash: replyHash, object: object, error: error, timestamp: Date.now()}; 
+            
+
+            this.streamSource.ingest(item);
+            
+        } else {
+            ObjectDiscoveryAgent.log.debug('Error validating object discovery reply');
         }
     }
 

src/mesh/agents/network/SecureNetworkAgent.ts

@@ -406,9 +406,9 @@ class SecureNetworkAgent implements Agent {
 
     secureForReceiving(connId: ConnectionId, localIdentity: Identity, timeout=DEFAULT_TIMEOUT) {
 
-        SecureNetworkAgent.logger.trace('Asked to verify ' + connId + ' for receiving with ' + localIdentity.hash());
+        SecureNetworkAgent.logger.trace('Asked to verify ' + connId + ' for receiving with ' + localIdentity.getLastHash());
 
-        const identityHash = localIdentity.hash();
+        const identityHash = localIdentity.getLastHash();
 
         let secured = this.getOrCreateConnectionSecuredForReceiving(connId, identityHash);
         secured.identity = localIdentity;
@@ -799,7 +799,7 @@ class SecureNetworkAgent implements Agent {
     private sendIdentity(connId: ConnectionId, identity: Identity, identityHash?: Hash) {
 
         if (identityHash === undefined) {
-            identityHash = identity.hash();
+            identityHash = identity.getLastHash();
         }
 
         let content: SendIdentityMessage = {

src/storage/store/Store.ts

@@ -515,7 +515,7 @@ class Store {
             }*/
         }
 
-        //console.log('+++ loaded ' + hash, object?.getClassName(), new Error().stack);
+        //console.log('+++ loaded ' + hash, object?.getClassName());
 
         if (object instanceof MutationOp) {
             //console.log('+++ target: ' + object.getTargetObject().getLastHash() + ' (' + object.getTargetObject().getClassName() + ')');

src/util/events.ts

@@ -15,7 +15,7 @@ import { Logger, LogLevel } from 'util/logging';
 
 
 type hash = string;
-type hashable = { hash(): hash; };
+type hashable = { getLastHash(): hash; };
 
 type location<T> = {name: string, emitter: T};
 //type path<T> = Array<location<T>>;
@@ -46,7 +46,7 @@ class EventRelay<T extends hashable> {
     constructor(source: T, upstreamRelays: Map<string, EventRelay<T>>) {
 
         this.emitter     = source;
-        this.emitterHash = source.hash(); 
+        this.emitterHash = source.getLastHash(); 
 
         this.upstreamRelays = new Map();