Merge remote-tracking branch 'origin/candidate-5.0.4'

Conflicts:
	thorlcr/activities/hashdistrib/thhashdistribslave.cpp
	version.cmake

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

docs/ECLWatch/TheECLWatchMan.xml

@@ -39,7 +39,7 @@
       similarity to actual persons, living or dead, is purely
       coincidental.</para>
 
-      <para></para>
+      <para> </para>
     </legalnotice>
 
     <xi:include href="common/Version.xml" xpointer="FooterInfo"

docs/Installing_and_RunningTheHPCCPlatform/Inst-Mods/UserSecurityMaint.xml

@@ -186,7 +186,7 @@
         </orderedlist></para>
     </sect3>
 
-    <sect3>
+    <sect3 role="brk">
       <title>Setting and modifying user permissions</title>
 
       <para>Access to ECL Watch and its features is controlled using a login
@@ -403,6 +403,78 @@
       </sect4>
 
       <sect4>
+        <title>To promote a user to an Administrator</title>
+
+        <para>To modify a users credentials you must have Administrator level
+        access. To promote a user to an HPCC Administrator, add the user to
+        the <emphasis role="bold">Administrators</emphasis> group.</para>
+
+        <para>Click on the<emphasis role="bold"> Operations</emphasis> icon,
+        then click the <emphasis role="bold">Security</emphasis> link from the
+        navigation sub-menu. <graphic
+        fileref="../../images/Permissions001.jpg" /></para>
+
+        <orderedlist>
+          <listitem>
+            <para>Click on the <emphasis role="bold">Users<emphasis>
+            tab</emphasis></emphasis>.</para>
+
+            <para>The users display in a list.</para>
+          </listitem>
+
+          <listitem>
+            <para>Select the user (or users) to promote. Check the box next to
+            the Username to select.</para>
+
+            <para>This enables the Users action buttons.</para>
+          </listitem>
+
+          <listitem>
+            <!--NOTE: UPdate this image***-->
+
+            <para>Press the <emphasis role="bold">Open</emphasis> action
+            button. <graphic fileref="../../images/Perm008.jpg" /></para>
+
+            <para>A tab opens for each user selected. On that tab there are
+            three sub-tabs.</para>
+          </listitem>
+
+          <listitem>
+            <para>Click on the tab for the user to modify (if more than one
+            user selected, repeat for each user).</para>
+
+            <para>On the user's tab there are three sub-tabs.</para>
+
+            <para>Click on the <emphasis role="bold">Member Of</emphasis>
+            sub-tab. <graphic fileref="../../images/Perm016.jpg" /></para>
+          </listitem>
+
+          <listitem>
+            <para>Select <emphasis role="bold">Administrators</emphasis> by
+            placing a check in box.</para>
+
+            <para><variablelist>
+                <varlistentry>
+                  <term>NOTE:</term>
+
+                  <listitem>
+                    <para>The name of the default Administrator group could
+                    vary. For example, in Active Directory, it is
+                    "Administrators", in Open LDAP it is "Directory
+                    Administrators".</para>
+                  </listitem>
+                </varlistentry>
+              </variablelist></para>
+          </listitem>
+
+          <listitem>
+            <para>The changes are automatically saved. Close the
+            tab(s).</para>
+          </listitem>
+        </orderedlist>
+      </sect4>
+
+      <sect4>
         <title>To delete a user from a group:</title>
 
         <para>To delete a user you must have Administrator level

docs/Installing_and_RunningTheHPCCPlatform/Inst-Mods/hpcc_ldap.xml

@@ -203,7 +203,12 @@
 
     <para>This section contains the information to install and implement LDAP
     based authentication. LDAP Authentication provides the most options for
-    securing your system, or parts of your system.</para>
+    securing your system, or parts of your system. In addition to these
+    configuration settings you must run the <emphasis
+    role="bold">initldap</emphasis> utility to create the appropriate OUs and
+    the default HPCC Admin user on your LDAP server.</para>
+
+    <!--***Note: (9/2014) Adding Documentation for initLdap.*** -->
 
     <sect3>
       <title>Connect to Configuration Manager</title>
@@ -552,6 +557,35 @@
           </tbody>
         </tgroup>
       </informaltable></para>
+  </sect2>
+
+  <sect2>
+    <title>Installing the Default Admin user</title>
+
+    <para>After enabling your configuration for LDAP security, you need to run
+    the <emphasis role="bold">initldap</emphasis> utility that installs the
+    initial security components and the default users.</para>
+
+    <sect3>
+      <title>The initldap Utility</title>
+
+      <para>The initldap utility is used to create the HPCC Administrator's
+      user account and the HPCC OUs. The initldap utility uses the settings in
+      the LDAPServer component(s) in the environment.xml bound to the
+      configured ESPs.</para>
+
+      <para>Once you complete your configuration, (add the LDAPServer
+      components and distribute your environment.xml file) you must then run
+      the <emphasis role="bold">initldap</emphasis> utility.</para>
+
+      <programlisting>sudo /opt/HPCCSystems/bin/initldap</programlisting>
+
+      <para>The <emphasis role="bold">initldap</emphasis> utility prompts you
+      for LDAP Administrator credentials. Enter the appropriate values when
+      prompted. The initldap utility uses those values to initialize the LDAP
+      server and create the HPCC System User. It also creates all of the
+      required OUs.</para>
+    </sect3>
 
     <sect3>
       <title>Using the addScopes tool</title>
@@ -560,19 +594,19 @@
       “hpccinternal::&lt;user&gt;” file scope is also created granting new
       users full access to that scope and restricting access to other users.
       This file scope is used to store temporary HPCC files such as spill
-      files and temp files. </para>
+      files and temp files.</para>
 
       <para>If you are enabling LDAP file scope security and already have user
       accounts, you should run the addScopes utility program to create the
-      hpccinternal::&lt;user&gt; scope for those existing users. </para>
+      hpccinternal::&lt;user&gt; scope for those existing users.</para>
 
       <para>Users which already have this scope defined are ignored and so it
-      can be used on both new and legacy ESP user accounts safely. </para>
+      can be used on both new and legacy ESP user accounts safely.</para>
 
       <para>The tool is located in the <emphasis
       role="bold">/opt/HPCCSystems/bin/</emphasis> folder and to run it you
       must pass the location of <emphasis role="bold">daliconf.xml</emphasis>,
-      for example: </para>
+      for example:</para>
 
       <para><programlisting>/opt/HPCCSystems/bin/addScopes /var/lib/HPCCSystems/mydali/daliconf.xml</programlisting></para>
     </sect3>

roxie/ccd/ccdserver.cpp

@@ -24045,7 +24045,7 @@ public:
           puller(false),
           isLocal(_isLocal)
     {
-        variableIndexFileName = allFilesDynamic || factory->queryQueryFactory().isDynamic() || ((helper.getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename)) != 0);
+        variableIndexFileName = allFilesDynamic || factory->queryQueryFactory().isDynamic() || ((helper.getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename|JFindexfromactivity)) != 0);
         indexReadInputRecordVariable = indexReadMeta->isVariableSize();
         indexReadInput = NULL;
         rootIndex = NULL;
@@ -24871,7 +24871,7 @@ public:
           keySet(_keySet),
           translators(_translators)
     {
-        variableIndexFileName = allFilesDynamic || factory->queryQueryFactory().isDynamic() || ((helper.getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename)) != 0);
+        variableIndexFileName = allFilesDynamic || factory->queryQueryFactory().isDynamic() || ((helper.getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename|JFindexfromactivity)) != 0);
         indexReadInputRecordVariable = indexReadMeta->isVariableSize();
     }
 
@@ -25136,7 +25136,7 @@ public:
         enableFieldTranslation = queryFactory.queryOptions().enableFieldTranslation;
         translatorArray.setown(new TranslatorArray);
         joinFlags = helper->getJoinFlags();
-        variableIndexFileName = allFilesDynamic || _queryFactory.isDynamic() || ((joinFlags & (JFvarindexfilename|JFdynamicindexfilename)) != 0);
+        variableIndexFileName = allFilesDynamic || _queryFactory.isDynamic() || ((joinFlags & (JFvarindexfilename|JFdynamicindexfilename|JFindexfromactivity)) != 0);
         variableFetchFileName = allFilesDynamic || _queryFactory.isDynamic() || ((helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) != 0);
         if (!variableIndexFileName)
         {

system/security/LdapSecurity/ldapconnection.cpp

@@ -380,7 +380,12 @@ public:
             else if(m_serverType == ACTIVE_DIRECTORY)
                 m_sysuser_dn.append("cn=").append(m_sysuser_commonname.str()).append(",").append(m_sysuser_basedn.str());
             else if(m_serverType == OPEN_LDAP)
-                m_sysuser_dn.append("cn=").append(m_sysuser_commonname.str()).append(",").append(m_sysuser_basedn.str());
+            {
+                if (0==strcmp("Directory Manager",m_sysuser_commonname.str()))
+                    m_sysuser_dn.append("cn=").append(m_sysuser_commonname.str()).append(",").append(m_sysuser_basedn.str());
+                else
+                    m_sysuser_dn.append("uid=").append(m_sysuser_commonname.str()).append(",").append(m_sysuser_basedn.str()).append(",").append(m_basedn.str());
+            }
         }
 
         m_maxConnections = cfg->getPropInt(".//@maxConnections", DEFAULT_LDAP_POOL_SIZE);

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -188,6 +188,103 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
      */
     class CSender : public CSimpleInterface, implements IThreadFactory, implements IExceptionHandler
     {
+        class CTarget
+        {
+            CSender &owner;
+            unsigned target;
+            atomic_t activeWriters;
+            atomic_t senderFinished;
+            CSendBucketQueue pendingBuckets;
+            mutable CriticalSection crit;
+            Owned<CSendBucket> bucket;
+        public:
+            CTarget(CSender &_owner, unsigned _target) : owner(_owner), target(_target)
+            {
+                atomic_set(&activeWriters, 0);
+                atomic_set(&senderFinished, 0);
+            }
+            ~CTarget()
+            {
+                reset();
+            }
+            void reset()
+            {
+                loop
+                {
+                    CSendBucket *sendBucket = pendingBuckets.dequeueNow();
+                    if (!sendBucket)
+                        break;
+                    ::Release(sendBucket);
+                }
+                bucket.clear();
+                atomic_set(&activeWriters, 0);
+                atomic_set(&senderFinished, 0);
+            }
+            void send(CMessageBuffer &mb)
+            {
+                CriticalBlock b(crit);
+                if (!atomic_read(&senderFinished))
+                {
+                    if (owner.selfPush(target))
+                        assertex(target != owner.self);
+                    if (!owner.sendBlock(target, mb))
+                        atomic_set(&senderFinished, 1);
+                }
+            }
+            inline unsigned getNumPendingBuckets() const
+            {
+                return pendingBuckets.ordinality();
+            }
+            inline CSendBucket *dequeuePendingBucket()
+            {
+                return pendingBuckets.dequeueNow();
+            }
+            inline void enqueuePendingBucket(CSendBucket *bucket)
+            {
+                pendingBuckets.enqueue(bucket);
+            }
+            inline void incActiveWriters()
+            {
+                atomic_inc(&activeWriters);
+                ++owner.totalActiveWriters; // NB: incActiveWriters() is always called within a activeWritersLock crit
+            }
+            inline void decActiveWriters()
+            {
+                atomic_dec(&activeWriters);
+                --owner.totalActiveWriters; // NB: decActiveWriters() is always called within a activeWritersLock crit
+            }
+            inline unsigned getActiveWriters() const
+            {
+                return atomic_read(&activeWriters);
+            }
+            inline bool getSenderFinished() const
+            {
+                return atomic_read(&senderFinished);
+            }
+            inline void checkSenderFinished()
+            {
+                CriticalBlock b(crit);
+                if (!atomic_read(&senderFinished))
+                {
+                    atomic_set(&senderFinished, 1);
+                    atomic_inc(&owner.numFinished);
+                }
+            }
+            inline CSendBucket *queryBucket()
+            {
+                return bucket;
+            }
+            inline CSendBucket *queryBucketCreate()
+            {
+                if (!bucket)
+                    bucket.setown(new CSendBucket(owner.owner, target));
+                return bucket;
+            }
+            inline CSendBucket *getBucketClear()
+            {
+                return bucket.getClear();
+            }
+        };
         /*
          * CWriterHandler, a per thread class and member of the writerPool
          * a write handler, is given an initial CSendBucket and handles the dedup(if applicable)
@@ -203,21 +300,21 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
             CDistributorBase &distributor;
             Owned<CSendBucket> _sendBucket;
             unsigned nextPending;
-            bool aborted;
+            CTarget *target;
 
         public:
             IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
             CWriteHandler(CSender &_owner) : owner(_owner), distributor(_owner.owner)
             {
-                aborted = false;
+                target = NULL;
             }
             void init(void *startInfo)
             {
                 nextPending = getRandom()%distributor.numnodes;
                 _sendBucket.setown((CSendBucket *)startInfo);
-                owner.setActiveWriter(_sendBucket->queryDestination(), this);
-                aborted = false;
+                target = owner.targets.item(_sendBucket->queryDestination());
+                target->incActiveWriters();
             }
             void main()
             {
@@ -226,7 +323,7 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                 size32_t writerTotalSz = 0;
                 size32_t sendSz = 0;
                 MemoryBuffer mb;
-                while (!aborted)
+                while (!owner.aborted)
                 {
                     writerTotalSz += sendBucket->querySize();
                     owner.dedup(sendBucket); // conditional
@@ -243,7 +340,7 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                             // more added to dest I'm processing?
                             {
                                 CriticalBlock b(owner.activeWritersLock);
-                                sendBucket.setown(owner.pendingBuckets.item(dest)->dequeueNow());
+                                sendBucket.setown(target->dequeuePendingBucket());
                             }
                             if (sendBucket)
                             {
@@ -252,13 +349,13 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                                 continue; // NB: it will flow into else "remote" arm
                             }
                         }
-                        while (!aborted)
+                        while (!owner.aborted)
                         {
                             // JCSMORE check if worth compressing
                             CMessageBuffer msg;
                             fastLZCompressToBuffer(msg, mb.length(), mb.bufferBase());
                             mb.clear();
-                            owner.send(dest, msg);
+                            target->send(msg);
                             sendSz = 0;
                             if (wholeBucket)
                                 break;
@@ -270,25 +367,25 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                     // see if others to process
                     // NB: this will never start processing a bucket for a destination which already has an active writer.
                     CriticalBlock b(owner.activeWritersLock);
-                    owner.setActiveWriter(dest, NULL);
+                    target->decActiveWriters();
                     sendBucket.setown(owner.getAnotherBucket(nextPending));
                     if (!sendBucket)
+                    {
+                        target = NULL; // will be reinitialized to new target in init(), when thread pool thread is reused
                         break;
+                    }
                     dest = sendBucket->queryDestination();
-                    owner.setActiveWriter(dest, this);
+                    target = owner.targets.item(dest);
+                    target->incActiveWriters();
                     HDSendPrintLog3("CWriteHandler, now dealing with (b=%d), size=%d", sendBucket->queryDestination(), sendBucket->querySize());
                 }
             }
             bool canReuse() { return true; }
             bool stop() { return true; }
-            void abort()
-            {
-                aborted = true;
-            }
-        } **activeWriters;
+        };
 
         CDistributorBase &owner;
-        CriticalSection activeWritersLock;
+        mutable CriticalSection activeWritersLock;
         mutable SpinLock totalSzLock;
         SpinLock doDedupLock;
         IPointerArrayOf<CSendBucket> buckets;
@@ -297,52 +394,39 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
         bool senderFull, doDedup, aborted, initialized;
         Semaphore senderFullSem;
         Linked<IException> exception;
-        OwnedMalloc<bool> senderFinished;
-        unsigned numFinished, dedupSamples, dedupSuccesses, self;
+        atomic_t numFinished;
+        unsigned dedupSamples, dedupSuccesses, self;
         Owned<IThreadPool> writerPool;
-        PointerArrayOf<CSendBucketQueue> pendingBuckets;
-        unsigned numActiveWriters;
+        unsigned totalActiveWriters;
+        PointerArrayOf<CTarget> targets;
 
         void init()
         {
-            unsigned n;
-            for (unsigned n=0; n<owner.numnodes; n++)
-                buckets.append(NULL);
             totalSz = 0;
             senderFull = false;
-            senderFinished.allocateN(owner.numnodes, true);
-            numFinished = 0;
+            atomic_set(&numFinished, 0);
             dedupSamples = dedupSuccesses = 0;
             doDedup = owner.doDedup;
             writerPool.setown(createThreadPool("HashDist writer pool", this, this, owner.writerPoolSize, 5*60*1000));
             self = owner.activity->queryJob().queryMyRank()-1;
-            for (n=0; n<owner.numnodes; n++)
-                pendingBuckets.append(new CSendBucketQueue);
-            numActiveWriters = 0;
+
+            targets.ensure(owner.numnodes);
+            for (unsigned n=0; n<owner.numnodes; n++)
+                targets.append(new CTarget(*this, n));
+
+            totalActiveWriters = 0;
             aborted = false;
-            activeWriters = new CWriteHandler *[owner.numnodes];
-            memset(activeWriters, 0, owner.numnodes * sizeof(CWriteHandler *));
             initialized = true;
         }
         void reset()
         {
-            assertex(0 == numActiveWriters);
+            assertex(0 == totalActiveWriters);
             // unless it was aborted, there shouldn't be any pending or non-null buckets
             for (unsigned n=0; n<owner.numnodes; n++)
-            {
-                CSendBucketQueue *queue = pendingBuckets.item(n);
-                loop
-                {
-                    CSendBucket *bucket = queue->dequeueNow();
-                    if (!bucket)
-                        break;
-                    ::Release(bucket);
-                }
-                buckets.replace(NULL, n);
-            }
+                targets.item(n)->reset();
             totalSz = 0;
             senderFull = false;
-            numFinished = 0;
+            atomic_set(&numFinished, 0);
             aborted = false;
         }
         void reinit()
@@ -352,6 +436,11 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
             else
                 init();
         }
+        unsigned queryInactiveWriters() const
+        {
+            CriticalBlock b(activeWritersLock);
+            return owner.writerPoolSize - totalActiveWriters;
+        }
         void dedup(CSendBucket *sendBucket)
         {
             {
@@ -380,42 +469,6 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                 }
             }
         }
-        CSendBucket *queryBucket(unsigned n)
-        {
-            return buckets.item(n);
-        }
-        CSendBucket *queryBucketCreate(unsigned n)
-        {
-            CSendBucket *bucket = buckets.item(n);
-            if (!bucket)
-            {
-                bucket = new CSendBucket(owner, n);
-                buckets.replace(bucket, n);
-            }
-            return bucket;
-        }
-        CSendBucket *getBucketClear(unsigned n)
-        {
-            Linked<CSendBucket> bucket = buckets.item(n);
-            if (!bucket)
-                return NULL;
-            buckets.replace(NULL, n);
-            return bucket.getClear();
-        }
-        void send(unsigned dest, CMessageBuffer &mb)
-        {
-            if (!senderFinished[dest])
-            {
-                if (selfPush(dest))
-                    assertex(dest != self);
-                if (!owner.sendBlock(dest, mb))
-                {
-                    ActPrintLog(owner.activity, "CDistributorBase::sendBlock stopped slave %d", dest+1);
-                    senderFinished[dest] = true;
-                    numFinished++;
-                }
-            }
-        }
         void decTotal(size32_t sz)
         {
             SpinBlock b(totalSzLock);
@@ -432,7 +485,15 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
             SpinBlock b(totalSzLock);
             return totalSz;
         }
-        inline bool selfPush(unsigned i)
+        inline bool sendBlock(unsigned i, CMessageBuffer &msg)
+        {
+            if (owner.sendBlock(i, msg))
+                return true;
+            atomic_inc(&numFinished);
+            ActPrintLog(owner.activity, "CSender::sendBlock stopped slave %d (finished=%d)", i+1, atomic_read(&numFinished));
+            return false;
+        }
+        inline bool selfPush(unsigned i) const
         {
             return (i==self)&&!owner.pull;
         }
@@ -447,7 +508,7 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                     try
                     {
                         nullMsg.clear();
-                        owner.sendBlock(i, nullMsg);
+                        sendBlock(i, nullMsg);
                     }
                     catch (IException *e)
                     {
@@ -469,19 +530,8 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
         {
             if (initialized)
             {
-                delete [] activeWriters;
                 for (unsigned n=0; n<owner.numnodes; n++)
-                {
-                    CSendBucketQueue *queue = pendingBuckets.item(n);
-                    loop
-                    {
-                        CSendBucket *bucket = queue->dequeueNow();
-                        if (!bucket)
-                            break;
-                        ::Release(bucket);
-                    }
-                    delete queue;
-                }
+                    delete targets.item(n);
             }
         }
         CSendBucket *getAnotherBucket(unsigned &next)
@@ -490,15 +540,15 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
             unsigned start = next;
             loop
             {
-                unsigned current=next;
-                unsigned c = pendingBuckets.item(current)->ordinality();
+                CTarget *target = targets.item(next);
+                unsigned c = target->getNumPendingBuckets();
                 ++next;
                 if (next>=owner.numnodes)
                     next = 0;
                 if (c)
                 {
-                    if (NULL == activeWriters[current])
-                        return pendingBuckets.item(current)->dequeueNow();
+                    if (!owner.targetWriterLimit || (target->getActiveWriters() < owner.targetWriterLimit))
+                        return target->dequeuePendingBucket();
                 }
                 if (next == start)
                     return NULL;
@@ -506,13 +556,11 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
         }
         void add(CSendBucket *bucket)
         {
-            if (owner.selfstopped && !senderFinished[self])
-            {
-                senderFinished[self] = true;
-                ++numFinished;
-            }
+            if (owner.selfstopped)
+                targets.item(self)->checkSenderFinished();
             unsigned dest = bucket->queryDestination();
-            if (senderFinished[dest])
+            CTarget *target = targets.item(dest);
+            if (target->getSenderFinished())
             {
                 HDSendPrintLog2("CSender::add disposing of bucket [finished(%d)]", dest);
                 bucket->Release();
@@ -520,71 +568,14 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
             else
             {
                 CriticalBlock b(activeWritersLock);
-                CWriteHandler *writer = activeWriters[dest];
-                if (!writer && (numActiveWriters < owner.writerPoolSize))
+                if ((totalActiveWriters < owner.writerPoolSize) && (!owner.targetWriterLimit || (target->getActiveWriters() < owner.targetWriterLimit)))
                 {
-                    HDSendPrintLog2("CSender::add (new thread), dest=%d", dest);
+                    HDSendPrintLog3("CSender::add (new thread), dest=%d, active=%d", dest, totalActiveWriters);
                     writerPool->start(bucket);
                 }
                 else // an existing writer will pick up
-                    pendingBuckets.item(dest)->enqueue(bucket);
-            }
-        }
-        void setActiveWriter(unsigned n, CWriteHandler *writer)
-        {
-            if (writer)
-            {
-                assertex(!activeWriters[n]);
-                ++numActiveWriters;
-            }
-            else
-            {
-                assertex(activeWriters[n]);
-                --numActiveWriters;
-            }
-            activeWriters[n] = writer;
-        }
-        unsigned getSendCandidate(bool &doSelf)
-        {
-            unsigned i;
-            unsigned maxsz=0;
-            for (i=0;i<owner.numnodes;i++)
-            {
-                CSendBucket *bucket = queryBucket(i);
-                if (bucket)
-                {
-                    size32_t bucketSz = bucket->querySize();
-                    if (bucketSz > maxsz)
-                        maxsz = bucketSz;
-                }
-            }
-            doSelf = false;
-            if (0 == maxsz)
-                return NotFound;
-            candidates.kill();
-            for (i=0; i<owner.numnodes; i++)
-            {
-                CSendBucket *bucket = queryBucket(i);
-                if (bucket)
-                {
-                    size32_t bucketSz = bucket->querySize();
-                    if (bucketSz > maxsz/2)
-                    {
-                        if (i==self)
-                            doSelf = true;
-                        else
-                            candidates.append(i);
-                    }
-                }
+                    target->enqueuePendingBucket(bucket);
             }
-            if (0 == candidates.ordinality())
-                return NotFound;
-            unsigned h;
-            if (candidates.ordinality()==1)
-                h = candidates.item(0);
-            else
-                h = candidates.item(getRandom()%candidates.ordinality());
-            return h;
         }
         void process(IRowStream *input)
         {
@@ -594,7 +585,7 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
             rowcount_t totalSent = 0;
             try
             {
-                while (!aborted && numFinished < owner.numnodes)
+                while (!aborted && atomic_read(&numFinished) < owner.numnodes)
                 {
                     while (queryTotalSz() >= owner.inputBufferSize)
                     {
@@ -602,40 +593,110 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                             break;
 
                         HDSendPrintLog("process exceeded inputBufferSize");
-                        bool doSelf;
-                        unsigned which = getSendCandidate(doSelf);
-                        if (NotFound != which)
-                        {
-                            HDSendPrintLog3("process exceeded: send to %d, size=%d", which, queryBucket(which)->querySize());
-                            add(getBucketClear(which));
-                        }
-                        if (doSelf)
+
+                        // establish largest partial bucket
+                        unsigned maxSz=0;
+                        if (queryInactiveWriters())
                         {
-                            HDSendPrintLog2("process exceeded: doSelf, size=%d", queryBucket(self)->querySize());
-                            add(getBucketClear(self));
+                            for (unsigned i=0; i<owner.numnodes; i++)
+                            {
+                                CSendBucket *bucket = targets.item(i)->queryBucket();
+                                if (bucket)
+                                {
+                                    size32_t bucketSz = bucket->querySize();
+                                    if (bucketSz > maxSz)
+                                        maxSz = bucketSz;
+                                    HDSendPrintLog4("b[%d], rows=%d, size=%d", i, bucket->count(), bucketSz);
+                                }
+                            }
                         }
-                        else if (NotFound == which) // i.e. none
+                        /* Only add buckets if some inactive writers
+                         * choose larger candidate buckets to targets that are inactive
+                         * and randomize from that list which are queued to writers
+                         */
+                        if (maxSz)
                         {
-                            HDSendPrintLog("process exceeded inputBufferSize, none to send");
+                            // pick candidates that are at >= 50% size of largest
+                            candidates.clear();
+                            bool doSelf = false;
+                            unsigned inactiveWriters = queryInactiveWriters();
+                            for (unsigned i=0; i<owner.numnodes; i++)
                             {
-                                SpinBlock b(totalSzLock);
-                                // some may have been freed after lock
-                                if (totalSz < owner.inputBufferSize)
-                                    break;
-                                senderFull = true;
+                                CTarget *target = targets.item(i);
+                                CSendBucket *bucket = target->queryBucket();
+                                if (bucket)
+                                {
+                                    size32_t bucketSz = bucket->querySize();
+                                    if (bucketSz >= maxSz/2)
+                                    {
+                                        if (0 == target->getActiveWriters()) // only if there are no active writer threads for this target
+                                        {
+                                            if (i==self)
+                                                doSelf = true; // always send to self if candidate
+                                            else
+                                            {
+                                                candidates.append(i);
+                                                HDSendPrintLog4("c[%d], rows=%d, size=%d", i, bucket->count(), bucketSz);
+                                                /* NB: in theory could be more if some finished since checking, but that's okay
+                                                 * some candidates, or free space will be picked up in next section
+                                                 */
+                                                if (candidates.ordinality() >= inactiveWriters)
+                                                    break;
+                                            }
+                                        }
+                                    }
+                                }
                             }
-                            loop
+                            unsigned limit = owner.candidateLimit;
+                            while (candidates.ordinality())
                             {
-                                if (timer.elapsedCycles() >= queryOneSecCycles()*10)
-                                    ActPrintLog(owner.activity, "HD sender, waiting for space");
-                                timer.reset();
-
-                                if (senderFullSem.wait(10000))
-                                    break;
-                                if (aborted)
+                                if (0 == queryInactiveWriters())
                                     break;
+                                else
+                                {
+                                    unsigned pos = getRandom()%candidates.ordinality();
+                                    unsigned c = candidates.item(pos);
+                                    CTarget *target = targets.item(c);
+                                    CSendBucket *bucket = target->queryBucket();
+                                    assertex(bucket);
+                                    HDSendPrintLog3("process exceeded: send to %d, size=%d", c, bucket->querySize());
+                                    add(target->getBucketClear());
+                                    if (limit)
+                                    {
+                                        --limit;
+                                        if (0 == limit)
+                                            break;
+                                    }
+                                    candidates.remove(pos);
+                                }
+                            }
+                            if (doSelf)
+                            {
+                                CTarget *target = targets.item(self);
+                                CSendBucket *bucket = target->queryBucket();
+                                assertex(bucket);
+                                HDSendPrintLog2("process exceeded: doSelf, size=%d", bucket->querySize());
+                                add(target->getBucketClear());
                             }
                         }
+                        {
+                            SpinBlock b(totalSzLock);
+                            // some may have been written by now
+                            if (totalSz < owner.inputBufferSize)
+                                break;
+                            senderFull = true;
+                        }
+                        loop
+                        {
+                            if (timer.elapsedCycles() >= queryOneSecCycles()*10)
+                                ActPrintLog(owner.activity, "HD sender, waiting for space, active writers = %d", queryInactiveWriters());
+                            timer.reset();
+
+                            if (senderFullSem.wait(10000))
+                                break;
+                            if (aborted)
+                                break;
+                        }
                     }
                     if (aborted)
                         break;
@@ -643,11 +704,12 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                     if (!row)
                         break;
                     unsigned dest = owner.ihash->hash(row)%owner.numnodes;
-                    if (senderFinished[dest]) // does this need to be thread safe?
+                    CTarget *target = targets.item(dest);
+                    if (target->getSenderFinished()) // does this need to be thread safe?
                         ReleaseThorRow(row);
                     else
                     {
-                        CSendBucket *bucket = queryBucketCreate(dest);
+                        CSendBucket *bucket = target->queryBucketCreate();
                         size32_t rs;
                         bucket->add(row, rs);
                         totalSent++;
@@ -658,7 +720,7 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                         if (bucket->querySize() >= owner.bucketSendSize)
                         {
                             HDSendPrintLog3("adding new bucket: %d, size = %d", bucket->queryDestination(), bucket->querySize());
-                            add(getBucketClear(dest));
+                            add(target->getBucketClear());
                         }
                     }
                 }
@@ -678,7 +740,7 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                 ForEach(*iter)
                 {
                     unsigned dest=iter->get();
-                    Owned<CSendBucket> bucket = getBucketClear(dest);
+                    Owned<CSendBucket> bucket = targets.item(dest)->getBucketClear();
                     HDSendPrintLog4("Looking at last bucket(%d): %d, size = %d", dest, bucket.get()?bucket->queryDestination():0, bucket.get()?bucket->querySize():-1);
                     if (bucket && bucket->querySize())
                     {
@@ -700,16 +762,6 @@ class CDistributorBase : public CSimpleInterface, implements IHashDistributor, i
                 return;
             aborted = true;
             senderFullSem.signal();
-            if (initialized)
-            {
-                CriticalBlock b(activeWritersLock);
-                for (unsigned w=0; w<owner.numnodes; w++)
-                {
-                    CWriteHandler *writer = activeWriters[w];
-                    if (writer)
-                        writer->abort();
-                }
-            }
         }
     // IThreadFactory impl.
         virtual IPooledThread *createNew()
@@ -798,6 +850,8 @@ protected:
     CriticalSection putsect;
     bool pull, aborted;
     CSender sender;
+    unsigned candidateLimit;
+    unsigned targetWriterLimit;
 public:
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
@@ -811,9 +865,9 @@ public:
         iCompare = NULL;
         ihash = NULL;
         fixedEstSize = 0;
-        bucketSendSize = globals->getPropInt("@hd_out_buffer_size", DEFAULT_OUT_BUFFER_SIZE);
+        bucketSendSize = activity->getOptUInt(THOROPT_HDIST_BUCKET_SIZE, DEFAULT_OUT_BUFFER_SIZE);
         istop = _istop;
-        inputBufferSize = globals->getPropInt("@hd_in_buffer_size", DEFAULT_IN_BUFFER_SIZE);
+        inputBufferSize = activity->getOptUInt(THOROPT_HDIST_BUFFER_SIZE, DEFAULT_IN_BUFFER_SIZE);
         pullBufferSize = DISTRIBUTE_PULL_BUFFER_SIZE;
         selfstopped = false;
         pull = false;
@@ -823,9 +877,14 @@ public:
         if (allowSpill)
             ActPrintLog(activity, "Using spilling buffer (will spill if overflows)");
         writerPoolSize = activity->getOptUInt(THOROPT_HDIST_WRITE_POOL_SIZE, DEFAULT_WRITEPOOLSIZE);
-        if (writerPoolSize>numnodes)
-            writerPoolSize = numnodes; // no point in more
+        if (writerPoolSize>(numnodes*2))
+            writerPoolSize = numnodes*2; // limit to 2 per target
         ActPrintLog(activity, "Writer thread pool size : %d", writerPoolSize);
+        candidateLimit = activity->getOptUInt(THOROPT_HDIST_CANDIDATELIMIT);
+        ActPrintLog(activity, "candidateLimit : %d", candidateLimit);
+        ActPrintLog(activity, "inputBufferSize : %d, bucketSendSize = %d", inputBufferSize, bucketSendSize);
+        targetWriterLimit = activity->getOptUInt(THOROPT_HDIST_TARGETWRITELIMIT);
+        ActPrintLog(activity, "targetWriterLimit : %d", targetWriterLimit);
     }
 
     ~CDistributorBase()
@@ -1013,7 +1072,6 @@ public:
             piperd->stop();
         pipewr.clear();
         ActPrintLog(activity, "HDIST: Read loop done");
-
     }
 
     void sendloop()

thorlcr/thorutil/thormisc.hpp

@@ -50,6 +50,10 @@
 #define THOROPT_COMPRESS_SPILLS       "compressInternalSpills"  // Compress internal spills, e.g. spills created by lookahead or sort gathering  (default = true)
 #define THOROPT_HDIST_SPILL           "hdistSpill"              // Allow distribute receiver to spill to disk, rather than blocking              (default = true)
 #define THOROPT_HDIST_WRITE_POOL_SIZE "hdistSendPoolSize"       // Distribute send thread pool size                                              (default = 16)
+#define THOROPT_HDIST_BUCKET_SIZE     "hd_out_buffer_size"      // Distribute target bucket send size                                            (default = 1MB)
+#define THOROPT_HDIST_BUFFER_SIZE     "hd_in_buffer_size"       // Distribute send buffer size (for all targets)                                 (default = 32MB)
+#define THOROPT_HDIST_CANDIDATELIMIT  "hdCandidateLimit"        // Limits # of buckets to push to the writers when send buffer is full           (default = is 50% largest)
+#define THOROPT_HDIST_TARGETWRITELIMIT "hdTargetLimit"          // Limit # of writer threads working on a single target                          (default = unbound, but picks round-robin)
 #define THOROPT_SPLITTER_SPILL        "splitterSpill"           // Force splitters to spill or not, default is to adhere to helper setting       (default = -1)
 #define THOROPT_LOOP_MAX_EMPTY        "loopMaxEmpty"            // Max # of iterations that LOOP can cycle through with 0 results before errors  (default = 1000)
 #define THOROPT_SMALLSORT             "smallSortThreshold"      // Use minisort approach, if estimate size of data to sort is below this setting (default = 0)

tools/CMakeLists.txt

@@ -18,6 +18,7 @@ HPCC_ADD_SUBDIRECTORY (esdlcmd-xml)
 HPCC_ADD_SUBDIRECTORY (hidl)
 HPCC_ADD_SUBDIRECTORY (backupnode "PLATFORM")
 HPCC_ADD_SUBDIRECTORY (addScopes "PLATFORM")
+HPCC_ADD_SUBDIRECTORY (initldap "PLATFORM")
 HPCC_ADD_SUBDIRECTORY (combine "PLATFORM")
 HPCC_ADD_SUBDIRECTORY (dumpkey "PLATFORM")
 HPCC_ADD_SUBDIRECTORY (keydiff "PLATFORM")

tools/initldap/CMakeLists.txt

@@ -0,0 +1,54 @@
+################################################################################
+#    HPCC SYSTEMS software Copyright (C) 2014 HPCC Systems.
+#
+#    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.
+################################################################################
+
+# Component: initldap
+
+#####################################################
+# Description:
+# ------------
+#    Cmake Input File for initldap
+#####################################################
+
+
+project( initldap )
+
+set (    SRCS
+         initldap.cpp
+    )
+
+include_directories (
+         ./../../system/security/LdapSecurity
+         ./../../system/security/shared
+         ./../../system/jlib
+         ./../../system/include
+         ${CMAKE_BINARY_DIR}
+         ${CMAKE_BINARY_DIR}/oss
+    )
+
+ADD_DEFINITIONS( -D_CONSOLE )
+
+
+HPCC_ADD_EXECUTABLE ( initldap ${SRCS} )
+install ( TARGETS initldap RUNTIME DESTINATION ${EXEC_DIR} )
+
+
+
+target_link_libraries ( initldap
+         jlib
+         LdapSecurity
+    )
+
+

tools/initldap/initldap.cpp

@@ -0,0 +1,246 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2014 HPCC Systems.
+
+    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.
+############################################################################## */
+#include "ldapsecurity.ipp"
+#include "ldapsecurity.hpp"
+#include "build-config.h"
+
+#ifndef _WIN32
+#include <unistd.h>
+#endif
+
+
+//-----------------------------------------------------
+//
+//-----------------------------------------------------
+void usage()
+{
+    fprintf(stdout, "\nUsage: initldap");
+    fprintf(stdout, "\n\n\tinitldap creates an initial HPCC Admin user account\n\tand all HPCC organization units, using the setting entered into configmanager 'LDAPServer' component\n");
+    fprintf(stdout, "\n");
+}
+
+//-----------------------------------------------------
+//
+//-----------------------------------------------------
+bool initLDAP(IPropertyTree * ldapProps)
+{
+    StringAttr serverType( ldapProps->queryProp("@serverType") );
+    if (!serverType.length())
+    {
+        fprintf(stderr, "\nERROR: serverType not set in LDAPServer component");
+        return false;
+    }
+
+    StringBuffer hpccUser;
+    StringBuffer hpccPwd;
+    ldapProps->getProp("@systemUser", hpccUser);
+    ldapProps->getProp("@systemPassword", hpccPwd);
+    if (0==hpccUser.length() || 0==hpccPwd.length())
+    {
+        fprintf(stderr, "\nERROR: HPCC systemUser credentials not found in configuration");
+        return false;
+    }
+
+    StringBuffer ldapAddress;
+    ldapProps->getProp("@ldapAddress", ldapAddress);
+
+    //Get LDAP admin creds from user
+    char buff[100];
+    fprintf(stdout, "\nEnter the '%s' LDAP Admin User name on '%s'...",serverType.get(),ldapAddress.str());
+    do
+    {
+        fgets(buff, sizeof(buff), stdin);
+    }
+    while (buff[0] == (char)'\n');
+
+    if (buff[strlen(buff)-1] == '\n')
+        buff[strlen(buff)-1] = (char)NULL;
+    StringAttr ldapUser(buff);
+
+    fprintf(stdout, "Enter the LDAP Admin user '%s' password...",ldapUser.get());
+    fgets(buff, sizeof(buff), stdin);
+    if (buff[strlen(buff)-1] == '\n')
+        buff[strlen(buff)-1] = (char)NULL;
+    StringAttr ldapPwd(buff);
+    if (0==ldapUser.length() || 0==ldapPwd.length())
+    {
+        fprintf(stderr, "\nERROR: Invalid LDAP Admin account credentials entered");
+        return false;
+    }
+
+    fprintf(stdout, "\nReady to initialize HPCC LDAP Environment, using the following settings");
+    fprintf(stdout, "\n\tLDAP Server     : %s", ldapAddress.str());
+    fprintf(stdout, "\n\tLDAP Type       : %s", serverType.get());
+    fprintf(stdout, "\n\tHPCC Admin User : %s", hpccUser.str());
+    fprintf(stdout, "\nProceed?  y/n ");
+    for (;;)
+    {
+        int c = getchar();
+        if (c == 'y' || c == 'Y')
+            break;
+        else if (c == 'n' || c == 'N')
+            return true;
+    }
+
+    if (stricmp(serverType.get(),"ActiveDirectory"))
+        ldapProps->setProp("@systemBasedn", "");
+
+    //Replace system user with LDAP Admin credentials
+    ldapProps->setProp("@systemUser", ldapUser);
+    ldapProps->setProp("@systemCommonName", ldapUser);
+    StringBuffer sb;
+    encrypt(sb,ldapPwd);
+    ldapProps->setProp("@systemPassword", sb.str());
+
+    //Create security manager. This creates the required OUs
+    Owned<ISecManager> secMgr;
+    try
+    {
+        secMgr.setown(newLdapSecManager("initldap", *LINK(ldapProps)));
+    }
+    catch(IException *e)
+    {
+        StringBuffer buff;
+        e->errorMessage(buff);
+        e->Release();
+        fprintf(stderr, "\nERROR: Unable to create security manager : %s", buff.str());
+        return false;
+    }
+
+    //Create HPCC Admin user
+    Owned<ISecUser> user = secMgr->createUser(hpccUser.str());
+    StringBuffer pwd;
+    decrypt(pwd, hpccPwd.str());
+    user->credentials().setPassword(pwd.str());
+    try { secMgr->addUser(*user.get()); }
+    catch(...) {}//user may already exist, so just move on
+
+    //Add HPCC admin user to Administrators group
+    CLdapSecManager* ldapSecMgr = dynamic_cast<CLdapSecManager*>(secMgr.get());
+    if (!ldapSecMgr)
+    {
+        fprintf(stderr, "\nERROR: Unable to access CLdapSecManager object");
+        return false;
+    }
+    StringAttr adminGroup;
+    bool isActiveDir = true;
+    if (0 == stricmp(serverType.get(),"ActiveDirectory"))
+        adminGroup.set("Administrators");
+    else
+        adminGroup.set("Directory Administrators");
+    try { ldapSecMgr->changeUserGroup("add", hpccUser.str(), adminGroup); }
+    catch(...) {}//user may already be in group so just move on
+
+    fprintf(stdout, "\n\nLDAP Initialization successful\n");
+    return true;
+}
+
+//-----------------------------------------------------
+//
+//-----------------------------------------------------
+int main(int argc, char* argv[])
+{
+#ifdef _NO_LDAP
+    fprintf(stderr, "System was built with _NO_LDAP\n");
+    return -1;
+#endif
+
+    for (int x = 1; x < argc; x++)
+    {
+        if (0==strncmp("-h", argv[x], 2))
+        {
+            usage();
+            exit(0);
+        }
+        else
+        {
+            fprintf(stderr, "\nERROR: Unrecognized parameter : '%s', enter 'initldap -h' for help\n", argv[x]);
+            exit(1);
+        }
+    }
+
+    InitModuleObjects();
+
+    //execute configgen to query the LDAP Server configuration(s)
+    StringBuffer cmd;
+    cmd.appendf("%s%cconfiggen -env %s%c%s -listldapservers", ADMIN_DIR,PATHSEPCHAR,CONFIG_DIR, PATHSEPCHAR, ENV_XML_FILE);
+
+    char * configBuffer = NULL;
+
+    //acquire LDAP configuration by executing configgen and capturing output
+    {
+        StringBuffer configBuff;
+        Owned<IPipeProcess> pipe = createPipeProcess();
+        if (pipe->run("configgen", cmd.str(), ".", false, true, true, 0))
+        {
+            Owned<ISimpleReadStream> pipeReader = pipe->getOutputStream();
+            const size32_t chunkSize = 8192;
+            for (;;)
+            {
+                size32_t sizeRead = pipeReader->read(chunkSize, configBuff.reserve(chunkSize));
+                if (sizeRead < chunkSize)
+                {
+                    configBuff.setLength(configBuff.length() - (chunkSize - sizeRead));
+                    break;
+                }
+            }
+            pipe->closeOutput();
+        }
+        int retcode = pipe->wait();
+        if (retcode)
+        {
+            fprintf(stderr, "\nERROR %d: unable to execute %s", retcode, cmd.str());
+            exit(1);
+        }
+        configBuffer = strdup(configBuff.str());
+    }
+
+    //Using the LDAP Server parms queried from configgen, build an
+    //LDAPSecurity property tree for each LDAP Server and call the LDAP
+    //Security Manager to create the needed entries
+    Owned<IPropertyTree> ldapProps;
+    char *saveptr;
+    char * pLine = strtok_r(configBuffer, "\n", &saveptr);
+    while (pLine)
+    {
+        if (pLine && 0==strcmp(pLine, "LDAPServerProcess"))
+        {
+            if (ldapProps)
+                initLDAP(ldapProps);
+            ldapProps.clear();
+            ldapProps.setown(createPTree("ldapSecurity"));
+        }
+        else
+        {
+            char * sep = strchr(pLine, ',');
+            if (sep)
+            {
+                *sep = (char)NULL;
+                ldapProps->addProp(pLine, sep+1);
+            }
+        }
+        pLine = strtok_r(NULL, "\n", &saveptr);
+    }
+    if (ldapProps)
+        initLDAP(ldapProps);
+    if (configBuffer)
+        free(configBuffer);
+    ldapProps.clear();
+
+    releaseAtoms();
+    return 0;
+}

tools/initldap/sourcedoc.xml

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+################################################################################
+#    HPCC SYSTEMS software Copyright (C) 2014 HPCC Systems.
+#
+#    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.
+################################################################################
+-->
+<!DOCTYPE section PUBLIC "-//OASIS//DTD DocBook XML V4.3//EN" "http://www.oasis-open.org/docbook/xml/4.3/docbookx.dtd">
+<section>
+    <title>tools/initldap</title>
+
+    <para>
+        The tools/initldap directory contains the sources for the tools/initldap tool.
+    </para>
+</section>

tools/sourcedoc.xml

@@ -38,4 +38,5 @@
     <xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="testsocket/sourcedoc.xml"/>
     <xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="vkey/sourcedoc.xml"/>
     <xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="wuget/sourcedoc.xml"/>
+    <xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="initldap/sourcedoc.xml"/>
 </section>