Merge remote-tracking branch 'cntk/master'

Common/BestGpu.cpp

@@ -11,7 +11,7 @@
 
 // From the SDK documentation:
 // "The NVML library can be found at: %ProgramW6432%\"NVIDIA Corporation"\NVSMI\ on Windows, but will not be added to the path. To dynamically link to NVML, add this path to the PATH environmental variable. To dynamically load NVML, call LoadLibrary with this path."
-// "On Linux the NVML library will be found on the standard library path. For 64-bit Linux, both the 32-bit and 64-bit NVML libraries will be installed.
+// "On Linux the NVML library will be found on the standard library path. For 64-bit Linux, both the 32-bit and 64-bit NVML libraries will be installed."
 
 #define _CRT_SECURE_NO_WARNINGS // "secure" CRT not available on all platforms  --add this at the top of all CPP files that give "function or variable may be unsafe" warnings
 #include "Platform.h"
@@ -39,16 +39,13 @@
 #define PATH_DELIMITER '/'
 #endif//__WINDOWS__
 #include <stdio.h>
+#include <string.h>
 
-#ifdef MPI_SUPPORT
-#include "mpi.h"
-#endif
 #include <memory>
-
 #include "CrossProcessMutex.h"
+#include "../../MachineLearning/CNTK/MPIWrapper.h"
+extern Microsoft::MSR::CNTK::MPIWrapper *g_mpi;
 
-extern int mpiRank;
-extern int mpiNumProcesses;
 
 // ---------------------------------------------------------------------------
 // BestGpu class
@@ -153,39 +150,8 @@ DEVICEID_TYPE DeviceFromConfig(const ConfigParameters& config)
 
     if (!_stricmp(val.c_str(), "Auto"))
     {
-#ifdef MPI_SUPPORT
-        // make sure deviceId is unique among processes on the same machine
-        g_bestGpu->AllowAll();
-        std::string MyName(getenv("COMPUTERNAME"));
-        for (int i = 0; i < mpiNumProcesses; i++)
-        {
-            DEVICEID_TYPE yourDeviceId = deviceId;
-            if (mpiRank == i)
-            {
-                std::vector<int> devices = g_bestGpu->GetDevices(1);
-                deviceId = yourDeviceId = (DEVICEID_TYPE)devices[0];
-            }
-            MPI_Bcast(&yourDeviceId, 1, MPI_INT, i, MPI_COMM_WORLD);
-            {
-                INT32 YourSize = (INT32)MyName.length();
-                MPI_Bcast(&YourSize,1,MPI_INT,i,MPI_COMM_WORLD);
-                vector<char> YourName(YourSize+1);
-                if (mpiRank == i)
-                    copy(MyName.begin(), MyName.end(), YourName.begin());
-                MPI_Bcast(YourName.data(), YourSize + 1, MPI_CHAR, i, MPI_COMM_WORLD);
-                if (mpiRank != i)
-                {
-                    if (!_strcmpi(MyName.data(), YourName.data()))
-                    {
-                        g_bestGpu->DisallowDevice(yourDeviceId);
-                    }
-                }
-            }
-        }
-#else
         deviceId = (DEVICEID_TYPE)
             g_bestGpu->GetDevice(BestGpuFlags(bLockGPU ? (bestGpuAvoidSharing | bestGpuExclusiveLock) : bestGpuAvoidSharing));
-#endif
     }
     else if (!_stricmp(val.c_str(), "All"))
     {

Common/DataReader.cpp

@@ -137,6 +137,37 @@ void DataReader<ElemType>::StartMinibatchLoop(size_t mbSize, size_t epoch, size_
         m_dataReader[m_ioNames[i]]->StartMinibatchLoop(mbSize, epoch, requestedEpochSamples);
 }
 
+//SupportsDistributedMBRead - Tells if the reader supports distributed minibatch reading for parallel training
+template<class ElemType>
+bool DataReader<ElemType>::SupportsDistributedMBRead() const
+{
+    bool supportsDistributedMBRead = true;
+    for (size_t i = 0; i < m_ioNames.size(); i++)
+    {
+        auto currReaderIter = m_dataReader.find(m_ioNames[i]);
+        assert(currReaderIter != m_dataReader.end());
+
+        supportsDistributedMBRead &= currReaderIter->second->SupportsDistributedMBRead();
+    }
+
+    return supportsDistributedMBRead;
+}
+
+//StartDistributedMinibatchLoop - Startup a distributed minibatch loop for parallel training
+// mbSize - [in] size of the minibatch (number of frames, etc.)
+// epoch - [in] epoch number for this loop
+// subsetNum - [in] the subset number of the current node in a group of parallel training nodes
+// numSubsets - [in] total number of nodes participating in the parallel training
+// requestedEpochSamples - [in] number of samples to randomize, defaults to requestDataSize which uses the number of samples there are in the dataset
+template<class ElemType>
+void DataReader<ElemType>::StartDistributedMinibatchLoop(size_t mbSize, size_t epoch, size_t subsetNum, size_t numSubsets, size_t requestedEpochSamples = requestDataSize)
+{
+    for (size_t i = 0; i < m_ioNames.size(); i++)
+    {
+        m_dataReader[m_ioNames[i]]->StartDistributedMinibatchLoop(mbSize, epoch, subsetNum, numSubsets, requestedEpochSamples);
+    }
+}
+
 // GetMinibatch - Get the next minibatch (features and labels)
 // matrices - [in] a map with named matrix types (i.e. 'features', 'labels') mapped to the corresponing matrix, 
 //             [out] each matrix resized if necessary containing data. 

Common/Include/DataReader.h

@@ -63,10 +63,23 @@ class DATAREADER_API IDataReader
     virtual void Init(const ConfigParameters& /*config*/) = 0;
     virtual void Destroy() = 0;
     virtual void StartMinibatchLoop(size_t mbSize, size_t epoch, size_t requestedEpochSamples=requestDataSize) = 0;
+
+    virtual bool SupportsDistributedMBRead() const { return false; };
+    virtual void StartDistributedMinibatchLoop(size_t mbSize, size_t epoch, size_t subsetNum, size_t numSubsets, size_t requestedEpochSamples = requestDataSize)
+    {
+        if (SupportsDistributedMBRead() || (numSubsets != 1) || (subsetNum != 0))
+        {
+            LogicError("This reader does not support distributed reading of mini-batches");
+        }
+
+        return StartMinibatchLoop(mbSize, epoch, requestedEpochSamples);
+    }
+
     virtual bool GetMinibatch(std::map<std::wstring, Matrix<ElemType>*>& matrices) = 0;
     virtual size_t NumberSlicesInEachRecurrentIter() = 0; 
     virtual int GetSentenceEndIdFromOutputLabel() { return -1; };
     virtual void SetNbrSlicesEachRecurrentIter(const size_t sz) { mBlgSize = sz; };
+    virtual bool RequireSentenceSeg() { return false; };
     virtual const std::map<LabelIdType, LabelType>& GetLabelMapping(const std::wstring&) { NOT_IMPLEMENTED; };
     virtual void SetLabelMapping(const std::wstring&, const std::map<LabelIdType, LabelType>&) { NOT_IMPLEMENTED; };
     virtual bool GetData(const std::wstring&, size_t, void*, size_t&, size_t) { NOT_IMPLEMENTED; };
@@ -174,6 +187,9 @@ class DataReader: public IDataReader<ElemType>, protected Plugin
     // requestedEpochSamples - [in] number of samples to randomize, defaults to requestDataSize which uses the number of samples there are in the dataset
     virtual void StartMinibatchLoop(size_t mbSize, size_t epoch, size_t requestedEpochSamples = requestDataSize);
 
+    virtual bool SupportsDistributedMBRead() const override;
+    virtual void StartDistributedMinibatchLoop(size_t mbSize, size_t epoch, size_t subsetNum, size_t numSubsets, size_t requestedEpochSamples = requestDataSize) override;
+
     // GetMinibatch - Get the next minibatch (features and labels)
     // matrices - [in] a map with named matrix types (i.e. 'features', 'labels') mapped to the corresponing matrix, 
     //             [out] each matrix resized if necessary containing data. 

Common/fileutil.cpp

@@ -1603,12 +1603,21 @@ void msra::files::make_intermediate_dirs (const wstring & filepath)
     wcscpy (&buf[0], filepath.c_str());
     wstring subpath;
     int skip = 0;
-    // if share (\\) then the first two levels (machine, share name) cannot be made
+#ifdef _WIN32
+    // On windows, if share (\\) then the first two levels (machine, share name) cannot be made
     if ((buf[0] == '/' && buf[1] == '/') || (buf[0] == '\\' && buf[1] == '\\'))
     {
         subpath = L"/";
         skip = 2;           // skip two levels (machine, share)
     }
+#else
+    // On unix, if the filepath starts with '/' then it is absolute 
+    // path and the created sub-paths should also start with '/'
+    if (buf[0] == '/')
+    {
+        subpath = L"/";
+    }
+#endif
     // make all constituents except the filename (to make a dir, include a trailing slash)
     wchar_t * context = nullptr;
     for (const wchar_t * p = wcstok_s (&buf[0], L"/\\", &context); p; p = wcstok_s (NULL, L"/\\", &context))