Merge pull request #4479 from richardkchapman/roxie-remote-lookup

Roxie remote lookup

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>

dali/base/dadfs.cpp

@@ -6069,13 +6069,25 @@ public:
 
 #define GROUP_CACHE_INTERVAL (1000*60)
 
+class CNamedGroupCacheEntry: public CInterface
+{
+public:
+    Linked<IGroup> group;
+    StringAttr name;
+    StringAttr groupdir;
+    unsigned cachedtime;
+
+    CNamedGroupCacheEntry(IGroup *_group, const char *_name, const char *_dir)
+    : group(_group), name(_name), groupdir(_dir)
+    {
+        cachedtime = msTick();
+    }
+};
+
 class CNamedGroupStore: public CInterface, implements INamedGroupStore
 {
     CriticalSection cachesect;
-    Owned<IGroup> cachedgroup;
-    StringAttr cachedname;
-    StringAttr cachedgroupdir;
-    unsigned cachedtime;
+    CIArrayOf<CNamedGroupCacheEntry> cache;
     unsigned defaultTimeout;
 
 public:
@@ -6084,7 +6096,6 @@ public:
     CNamedGroupStore()
     {
         defaultTimeout = INFINITE;
-        cachedtime = 0;
     }
 
     IGroup *dolookup(const char *logicalgroupname,IRemoteConnection *conn, StringBuffer *dirret)
@@ -6096,20 +6107,6 @@ public:
             return NULL;
         gname.toLowerCase();
         logicalgroupname = gname.str();
-        if ((gname.length()>9)&&(memcmp(logicalgroupname,"foreign::",9)==0)) {
-            StringBuffer eps;
-            const char *s = logicalgroupname+9;
-            while (*s&&((*s!=':')||(s[1]!=':')))
-                eps.append(*(s++));
-            if (*s) {
-                s+=2;
-                if (*s) {
-                    Owned<INode> dali = createINode(eps.str());
-                    if (dali) 
-                        return getRemoteGroup(dali,s,FOREIGN_DALI_TIMEOUT,dirret);
-                }
-            }
-        }
         bool isiprange = (*logicalgroupname!=0);
         for (const char *s1=logicalgroupname;*s1;s1++)
             if (isalpha(*s1)) {
@@ -6152,28 +6149,48 @@ public:
             logicalgroupname = gname.str();
         }
         StringAttr groupdir;
+        bool cached = false;
+        unsigned timeNow = msTick();
         {
             CriticalBlock block(cachesect);
-            if (cachedgroup.get()) {
-                if (msTick()-cachedtime>GROUP_CACHE_INTERVAL) {
-                    cachedgroup.clear();
-                    cachedname.clear();
-                    cachedgroupdir.clear();
+            ForEachItemInRev(idx, cache)
+            {
+                CNamedGroupCacheEntry &entry = cache.item(idx);
+                if (timeNow-entry.cachedtime > GROUP_CACHE_INTERVAL)
+                {
+                    cache.remove(idx);
                 }
-                else if (strcmp(gname.str(),cachedname.get())==0) {
-                    cachedtime = msTick();
-                    if (range.length()==0) {
+                else if (strcmp(gname.str(),entry.name.get())==0)
+                {
+                    cached = true;
+                    if (range.length()==0)
+                    {
                         if (dirret)
-                            dirret->append(cachedgroupdir);
-                        return cachedgroup.getLink();
+                            dirret->append(entry.groupdir);
+                        return entry.group.getLink();
                     }
                     // there is a range so copy to epa
-                    cachedgroup->getSocketEndpoints(epa);
-                    groupdir.set(cachedgroupdir);
+                    entry.group->getSocketEndpoints(epa);
+                    groupdir.set(entry.groupdir);
+                    break;
                 }
             }
         }
-        if (epa.ordinality()==0) {
+        if ((gname.length()>9)&&(memcmp(logicalgroupname,"foreign::",9)==0)) {
+            StringBuffer eps;
+            const char *s = logicalgroupname+9;
+            while (*s&&((*s!=':')||(s[1]!=':')))
+                eps.append(*(s++));
+            if (*s) {
+                s+=2;
+                if (*s) {
+                    Owned<INode> dali = createINode(eps.str());
+                    if (!dali || !getRemoteGroup(dali, s, FOREIGN_DALI_TIMEOUT, groupdir, epa))
+                        return NULL;
+                }
+            }
+        }
+        else if (epa.ordinality()==0) {
             struct sLock
             {
                 sLock()  { lock = NULL; };
@@ -6196,30 +6213,33 @@ public:
                 epa.append(ep);
             }
         }
-        IGroup *ret = createIGroup(epa);
+        Owned<IGroup> ret = createIGroup(epa);
+        if (!cached)
         {
             CriticalBlock block(cachesect);
-            cachedgroup.set(ret);
-            cachedname.set(gname);
-            cachedgroupdir.set(groupdir);
-            cachedtime = msTick();
+            cache.append(*new CNamedGroupCacheEntry(ret, gname, groupdir));
         }
-        if (range.length()) {
+        if (range.length())
+        {
             SocketEndpointArray epar;
             const char *s = range.str();
-            while (*s) {
+            while (*s)
+            {
                 unsigned start = 0;
-                while (isdigit(*s)) {
+                while (isdigit(*s))
+                {
                     start = start*10+*s-'0';
                     s++;
                 }
                 if (!start)
                     break;
                 unsigned end;
-                if (*s=='-') {
+                if (*s=='-')
+                {
                     s++;
                     end = 0;
-                    while (isdigit(*s)) {
+                    while (isdigit(*s))
+                    {
                         end = end*10+*s-'0';
                         s++;
                     }
@@ -6228,7 +6248,8 @@ public:
                 }
                 else 
                     end = start;
-                if ((start>epa.ordinality())||(end>epa.ordinality())) {
+                if ((start>epa.ordinality())||(end>epa.ordinality()))
+                {
                     s = range.str();
                     break;
                 }
@@ -6244,12 +6265,11 @@ public:
             }
             if (*s) 
                 throw MakeStringException(-1,"Invalid group range %s",range.str());
-            ::Release(ret);
-            ret = createIGroup(epar);
+            ret.setown(createIGroup(epar));
         }
         if (dirret)
             dirret->append(groupdir);
-        return ret;
+        return ret.getClear();
     }
 
     IGroup *lookup(const char *logicalgroupname)
@@ -6337,11 +6357,10 @@ public:
         connlock.conn->queryRoot()->removeProp(prop.str()); 
         doadd(connlock,name.str(),group,cluster,dir);
         {                                                           
-            CriticalBlock block(cachesect);                     
-            cachedgroup.set(group); // may be NULL
-            cachedname.set(name.str());
-            cachedgroupdir.set(dir);
-            cachedtime = msTick();
+            CriticalBlock block(cachesect);
+            cache.kill();
+            if (group)
+                cache.append(*new CNamedGroupCacheEntry(group, name.str(), dir));
         }
     }
 
@@ -6408,12 +6427,18 @@ public:
             }
         }
         CriticalBlock block(cachesect);
-        cachedgroup.clear();
-        cachedname.clear();
-        cachedgroupdir.clear();
+        cache.kill();
     }
 
-    IGroup *getRemoteGroup(const INode *foreigndali, const char *gname, unsigned foreigndalitimeout, StringBuffer *dirret)
+    unsigned setDefaultTimeout(unsigned timems)
+    {
+        unsigned ret = defaultTimeout;
+        defaultTimeout = timems;
+        return ret;
+    }
+
+private:
+    bool getRemoteGroup(const INode *foreigndali, const char *gname, unsigned foreigndalitimeout, StringAttr &groupdir, SocketEndpointArray &epa)
     {
         StringBuffer lcname(gname);
         gname = lcname.trim().toLowerCase().str();
@@ -6423,7 +6448,7 @@ public:
         foreignDaliSendRecv(foreigndali,mb,foreigndalitimeout);
         checkDfsReplyException(mb);
         if (mb.length()==0)
-            return NULL;
+            return false;
         byte ok;
         mb.read(ok);
         if (ok!=1) {
@@ -6432,39 +6457,21 @@ public:
                 mb.skip(mbsz-1);
                 mb.read(ok);
                 if (ok!=1) 
-                    return NULL;
+                    return false;
             }
             else
-                return NULL;
+                return false;
         }
         Owned<IPropertyTree> pt = createPTree(mb);
         Owned<IPropertyTreeIterator> pe = pt->getElements("Node");
-        SocketEndpointArray epa;
+        groupdir.set(pt->queryProp("@dir"));
         ForEach(*pe) {
             SocketEndpoint ep(pe->query().queryProp("@ip"));
             epa.append(ep);
         }
-        IGroup *ret = createIGroup(epa);
-        {
-            CriticalBlock block(cachesect);
-            cachedgroup.set(ret);
-            cachedname.set(gname);
-            cachedgroupdir.set(pt->queryProp("@dir"));
-            if (dirret)
-                dirret->append(cachedgroupdir);
-            cachedtime = msTick();
-        }
-        return ret;
+        return epa.ordinality() > 0;
     }
 
-    unsigned setDefaultTimeout(unsigned timems)
-    {
-        unsigned ret = defaultTimeout;
-        defaultTimeout = timems;
-        return ret;
-    }
-
-
 };
 
 static CNamedGroupStore *groupStore = NULL;

dali/base/dadfs.hpp

@@ -594,7 +594,6 @@ interface INamedGroupStore: implements IGroupResolver
     virtual bool find(IGroup *grp, StringBuffer &lname, bool add=false) = 0;
     virtual void addUnique(IGroup *group,StringBuffer &lname,const char *dir=NULL) = 0;
     virtual void swapNode(const IpAddress &from, const IpAddress &to) = 0;
-    virtual IGroup *getRemoteGroup(const INode *foreigndali, const char *gname, unsigned foreigndalitimeout=FOREIGN_DALI_TIMEOUT, StringBuffer *dir=NULL) = 0;
     virtual IGroup *lookup(const char *logicalgroupname, StringBuffer &dir) = 0;
     virtual unsigned setDefaultTimeout(unsigned timems) = 0;                                    // sets default timeout for SDS connections and locking                                                                                         // returns previous value
 

dali/dfu/dfuutil.cpp

@@ -402,11 +402,9 @@ class CFileCloner
             throw afor2.exc.getClear();
     }
 
-
-
     void cloneSubFile(IPropertyTree *ftree,const char *destfilename, INode *srcdali)   // name already has prefix added
     {
-        Owned<IFileDescriptor> srcfdesc = deserializeFileDescriptorTree(ftree,&queryNamedGroupStore(),0);
+        Owned<IFileDescriptor> srcfdesc = deserializeFileDescriptorTree(ftree, NULL, 0);
         const char * kind = srcfdesc->queryProperties().queryProp("@kind");
         bool iskey = kind&&(strcmp(kind,"key")==0);
 
@@ -437,11 +435,6 @@ class CFileCloner
         dstfdesc->addCluster(cluster1,grp1,spec);
         if (iskey&&!cluster2.isEmpty())
             dstfdesc->addCluster(cluster2,grp2,spec2);
-#ifdef _TESTING
-//      LOGFDESC("createFileClone",dstfdesc);
-#endif
-
-
 
         for (unsigned pn=0;pn<srcfdesc->numParts();pn++) {
             offset_t sz = srcfdesc->queryPart(pn)->queryProperties().getPropInt64("@size",-1);
@@ -461,6 +454,17 @@ class CFileCloner
         {
             StringBuffer s;
             dstfdesc->queryProperties().setProp("@cloneFrom", srcdali->endpoint().getUrlStr(s).str());
+            unsigned numClusters = srcfdesc->numClusters();
+            for (unsigned clusterNum = 0; clusterNum < numClusters; clusterNum++)
+            {
+                StringBuffer sourceGroup;
+                srcfdesc->getClusterGroupName(clusterNum, sourceGroup, NULL);
+                Owned<IPropertyTree> groupInfo = createPTree("cloneFromGroup");
+                groupInfo->setProp("@groupName", sourceGroup);
+                ClusterPartDiskMapSpec &spec = srcfdesc->queryPartDiskMapping(clusterNum);
+                spec.toProp(groupInfo);
+                dstfdesc->queryProperties().addPropTree("cloneFromGroup", groupInfo.getClear());
+            }
         }
 
         Owned<IDistributedFile> dstfile = fdir->createNew(dstfdesc);
@@ -552,7 +556,6 @@ public:
         }
     }
 
-
     void cloneSuperFile(const char *filename, CDfsLogicalFileName &dlfn)
     {
         level++;

dali/dfu/dfuutil.hpp

@@ -75,7 +75,6 @@ interface IDFUhelper: extends IInterface
         IPropertyTree *relationships,   // if not NULL, tree will have all relationships filled in
         IUserDescriptor *user
     ) = 0;
-
 };
 
 IDFUhelper *createIDFUhelper();

roxie/ccd/ccddali.cpp

@@ -274,6 +274,53 @@ private:
         return localTree.getClear();
     }
 
+    IFileDescriptor *recreateCloneSource(IFileDescriptor *srcfdesc, const char *destfilename)
+    {
+        Owned<IFileDescriptor> dstfdesc = createFileDescriptor(srcfdesc->getProperties());
+        // calculate dest dir
+
+        CDfsLogicalFileName dstlfn;
+        if (!dstlfn.setValidate(destfilename,true))
+            throw MakeStringException(-1,"Logical name %s invalid",destfilename);
+
+        StringBuffer dstpartmask;
+        getPartMask(dstpartmask,destfilename,srcfdesc->numParts());
+        dstfdesc->setPartMask(dstpartmask.str());
+        unsigned np = srcfdesc->numParts();
+        dstfdesc->setNumParts(srcfdesc->numParts());
+        DFD_OS os = (getPathSepChar(srcfdesc->queryDefaultDir())=='\\')?DFD_OSwindows:DFD_OSunix;
+        StringBuffer dir;
+        StringBuffer dstdir;
+        makePhysicalPartName(dstlfn.get(),0,0,dstdir,false,os,NULL);
+        dstfdesc->setDefaultDir(dstdir.str());
+
+        for (unsigned pn=0;pn<srcfdesc->numParts();pn++) {
+            offset_t sz = srcfdesc->queryPart(pn)->queryProperties().getPropInt64("@size",-1);
+            if (sz!=(offset_t)-1)
+                dstfdesc->queryPart(pn)->queryProperties().setPropInt64("@size",sz);
+            StringBuffer dates;
+            if (srcfdesc->queryPart(pn)->queryProperties().getProp("@modified",dates))
+                dstfdesc->queryPart(pn)->queryProperties().setProp("@modified",dates.str());
+        }
+
+        const char *cloneFrom = srcfdesc->queryProperties().queryProp("@cloneFrom");
+        Owned<IPropertyTreeIterator> groups = srcfdesc->queryProperties().getElements("cloneFromGroup");
+        ForEach(*groups)
+        {
+            IPropertyTree &elem = groups->query();
+            const char *groupName = elem.queryProp("@groupName");
+            StringBuffer foreignGroup("foreign::");
+            foreignGroup.append(cloneFrom).append("::").append(groupName);
+            Owned<IGroup> group = queryNamedGroupStore().lookup(foreignGroup);  // NOTE - this is cached by the named group store
+            ClusterPartDiskMapSpec dmSpec;
+            dmSpec.fromProp(&elem);
+            dstfdesc->addCluster(groupName, group, dmSpec);
+        }
+
+        return dstfdesc.getClear();
+    }
+
+
 public:
 
     IMPLEMENT_IINTERFACE;
@@ -344,30 +391,38 @@ public:
         return ret.getClear();
     }
 
-    IFileDescriptor *checkClonedFromRemote(const char *_lfn, IFileDescriptor *fdesc, bool cacheIt, bool writeAccess)
+    IFileDescriptor *checkClonedFromRemote(const char *_lfn, IFileDescriptor *fdesc, bool cacheIt)
     {
+        // NOTE - we rely on the fact that  queryNamedGroupStore().lookup caches results,to avoid excessive load on remote dali
         if (_lfn && !strnicmp(_lfn, "foreign", 7)) //if need to support dali hopping should add each remote location
             return NULL;
         if (!fdesc || !fdesc->queryProperties().hasProp("@cloneFrom"))
             return NULL;
-        SocketEndpoint cloneFrom;
-        cloneFrom.set(fdesc->queryProperties().queryProp("@cloneFrom"));
-        if (cloneFrom.isNull())
-            return NULL;
-        CDfsLogicalFileName lfn;
-        lfn.set(_lfn);
-        lfn.setForeign(cloneFrom, false);
-        if (!connected())
-            return resolveCachedLFN(lfn.get());
-        Owned<IDistributedFile> cloneFile = resolveLFN(lfn.get(), cacheIt, writeAccess);
-        if (cloneFile)
+        if (fdesc->queryProperties().hasProp("cloneFromGroup"))
+        {
+            return recreateCloneSource(fdesc, _lfn);
+        }
+        else // Legacy mode - recently cloned files should have the extra info
         {
-            Owned<IFileDescriptor> cloneFDesc = cloneFile->getFileDescriptor();
-            if (cloneFDesc->numParts()==fdesc->numParts())
-                return cloneFDesc.getClear();
+            SocketEndpoint cloneFrom;
+            cloneFrom.set(fdesc->queryProperties().queryProp("@cloneFrom"));
+            if (cloneFrom.isNull())
+                return NULL;
+            CDfsLogicalFileName lfn;
+            lfn.set(_lfn);
+            lfn.setForeign(cloneFrom, false);
+            if (!connected())
+                return resolveCachedLFN(lfn.get());
+            Owned<IDistributedFile> cloneFile = resolveLFN(lfn.get(), cacheIt, false);
+            if (cloneFile)
+            {
+                Owned<IFileDescriptor> cloneFDesc = cloneFile->getFileDescriptor();
+                if (cloneFDesc->numParts()==fdesc->numParts())
+                    return cloneFDesc.getClear();
 
-            StringBuffer s;
-            DBGLOG(ROXIE_MISMATCH, "File %s cloneFrom(%s) mismatch", _lfn, cloneFrom.getIpText(s).str());
+                StringBuffer s;
+                DBGLOG(ROXIE_MISMATCH, "File %s cloneFrom(%s) mismatch", _lfn, cloneFrom.getIpText(s).str());
+            }
         }
         return NULL;
     }

roxie/ccd/ccddali.hpp

@@ -39,7 +39,7 @@ interface IRoxieDaliHelper : extends IInterface
 {
     virtual void commitCache() = 0;
     virtual bool connected() const = 0;
-    virtual IFileDescriptor *checkClonedFromRemote(const char *id, IFileDescriptor *fdesc, bool cacheIt, bool writeAccess) = 0;
+    virtual IFileDescriptor *checkClonedFromRemote(const char *id, IFileDescriptor *fdesc, bool cacheIt) = 0;
     virtual IDistributedFile *resolveLFN(const char *filename, bool cacheIt, bool writeAccess) = 0;
     virtual IFileDescriptor *resolveCachedLFN(const char *filename) = 0;
     virtual IConstWorkUnit *attachWorkunit(const char *wuid, ILoadedDllEntry *source) = 0;

roxie/ccd/ccdfile.hpp

@@ -23,7 +23,7 @@
 #include "dautils.hpp"
 
 enum RoxieFileStatus { FileSizeMismatch, FileDateMismatch, FileCRCMismatch, FileIsValid, FileNotFound };
-enum RoxieFileType { ROXIE_WU_DLL, ROXIE_PLUGIN_DLL, ROXIE_KEY, ROXIE_FILE, ROXIE_PATCH, ROXIE_BASEINDEX };
+enum RoxieFileType { ROXIE_KEY, ROXIE_FILE };
 interface IFileIOArray;
 interface IRoxieFileCache;
 
@@ -43,15 +43,9 @@ interface ILazyFileIO : extends IFileIO
     virtual int getLinkCount() const = 0;
     virtual bool createHardFileLink() = 0;
 
-    virtual void setBaseIndexFileName(const char *val) =0;
-    virtual const char *queryBaseIndexFileName() = 0;
-    virtual void setPatchFile(ILazyFileIO *val) = 0;
-    virtual ILazyFileIO *queryPatchFile() = 0;
-
     virtual unsigned getLastAccessed() const = 0;
     virtual bool isOpen() const = 0;
     virtual void close() = 0;
-    virtual RoxieFileType getFileType() = 0;
     virtual void setCopying(bool copying) = 0;
     virtual bool isCopying() const = 0;
     virtual IMemoryMappedFile *queryMappedFile() = 0;
@@ -64,9 +58,8 @@ extern ILazyFileIO *createDynamicFile(const char *id, IPartDescriptor *pdesc, Ro
 
 interface IRoxieFileCache : extends IInterface
 {
-    virtual ILazyFileIO *lookupFile(const char *id, unsigned partNo, RoxieFileType fileType, const char *localLocation, const char *baseIndexFileName, ILazyFileIO *patchFile, const StringArray &peerRoxieCopiedLocationInfo, const StringArray &deployedLocationInfo, offset_t size, const CDateTime &modified, bool memFile, bool isRemote, bool startFileCopy, bool doForegroundCopy, unsigned crc, bool isCompressed, const char *lookupDali) = 0;
-    virtual IFileIO *lookupDllFile(const char* dllname, const char *localLocation, const StringArray &remoteNames, unsigned crc, bool isRemote) = 0;
-    virtual RoxieFileStatus fileUpToDate(IFile *f, RoxieFileType fileType, offset_t size, const CDateTime &modified, unsigned crc, const char* id, bool isCompressed) = 0;
+    virtual ILazyFileIO *lookupFile(const char *lfn, RoxieFileType fileType, IPartDescriptor *pdesc, unsigned numParts, const StringArray &deployedLocationInfo, bool startFileCopy) = 0;
+    virtual RoxieFileStatus fileUpToDate(IFile *f, offset_t size, const CDateTime &modified, unsigned crc, bool isCompressed) = 0;
     virtual int numFilesToCopy() = 0;
     virtual void closeExpired(bool remote) = 0;
     virtual StringAttrMapping *queryFileErrorList() = 0;  // returns list of files that could not be open
@@ -146,7 +139,5 @@ extern IRoxieWriteHandler *createRoxieWriteHandler(IRoxieDaliHelper *_daliHelper
 
 extern IRoxieFileCache &queryFileCache();
 extern IMemoryFile *createMemoryFile(const char *fileName);
-extern IDiffFileInfoCache *queryDiffFileInfoCache();
-extern void releaseDiffFileInfoCache();
 
 #endif

roxie/ccd/ccdmain.cpp

@@ -1080,7 +1080,6 @@ int STARTQUERY_API start_query(int argc, const char *argv[])
     cleanupPlugins();
     closeMulticastSockets();
     releaseSlaveDynamicFileCache();
-    releaseDiffFileInfoCache();
     releaseRoxieStateCache();
     setDaliServixSocketCaching(false);  // make sure it cleans up or you get bogus memleak reports
     setNodeCaching(false); // ditto

roxie/ccd/ccdstate.cpp

@@ -264,7 +264,7 @@ protected:
                 if (fd)
                 {
                     Owned <IResolvedFileCreator> result = createResolvedFile(fileName, NULL);
-                    Owned<IFileDescriptor> remoteFDesc = daliHelper->checkClonedFromRemote(fileName, fd, cacheIt, writeAccess);
+                    Owned<IFileDescriptor> remoteFDesc = daliHelper->checkClonedFromRemote(fileName, fd, cacheIt);
                     result->addSubFile(fd.getClear(), remoteFDesc.getClear());
                     return result.getClear();
                 }
@@ -1964,10 +1964,6 @@ private:
                     toXML(stats, reply);
                 }
             }
-            else if (stricmp(queryName, "control:queryDiffFileInfoCache")==0)
-            {
-                queryDiffFileInfoCache()->queryDiffFileNames(reply);
-            }
             else if (stricmp(queryName, "control:queryPackageInfo")==0)
             {
                 UNIMPLEMENTED;