Merge branch 'candidate-5.4.0'

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

common/remote/rmtssh.cpp

@@ -513,6 +513,8 @@ public:
             return;
         if (slaves.ordinality()>1) {
             PROGLOG("Results: (%d of %d finished)",done.ordinality(),slaves.ordinality());
+            int errCode = 0;
+            Owned<IMultiException> multiException = MakeMultiException();
             for (unsigned i=0;i<done.ordinality();i++) {
                 unsigned n = done.item(i);
                 StringBuffer res(replytext.item(n));
@@ -520,19 +522,30 @@ public:
                     res.setLength(res.length()-1);
                 if (res.length()==0)
                     PROGLOG("%d: %s(%d): [OK]",n+1,slaves.item(n),reply.item(n));
-                else if (strchr(res.str(),'\n')==NULL)
+                else if (strchr(res.str(),'\n')==NULL) {
                     PROGLOG("%d: %s(%d): %s",n+1,slaves.item(n),reply.item(n),res.str());
-                else
+                    if (reply.item(n)) {
+                        errCode = reply.item(n);
+                        multiException->append(*MakeStringExceptionDirect(reply.item(n),res.str()));
+                    }
+                }
+                else {
                     PROGLOG("%d: %s(%d):\n---------------------------\n%s\n===========================",n+1,slaves.item(n),reply.item(n),res.str());
+                    if (reply.item(n)) {
+                        errCode = reply.item(n);
+                        multiException->append(*MakeStringExceptionDirect(reply.item(n),res.str()));
+                    }
+                }
             }
+            if (errCode)
+                throw multiException.getClear();
         }
         else {
             StringBuffer res(replytext.item(0));
             while (res.length()&&(res.charAt(res.length()-1)<=' '))
                 res.setLength(res.length()-1);
             PROGLOG("%s result(%d):\n%s",useplink?"plink":"ssh",reply.item(0),res.str());
-            if (res.length())
-            {
+            if (res.length()) {
                 int code = reply.item(0);
                 if (code == 0)
                     code = -1;

deployment/deploy/DeployTask.cpp

@@ -333,6 +333,7 @@ public:
    bool transformFile(IXslProcessor& processor, IXslTransform& transform, const char* tempPath)
    {
      m_processed = true;
+     bool bDeleteFile = true;
 
      const char* xsl = getFileSpec(DT_SOURCE);
      const char* target = getFileSpec(DT_TARGET);
@@ -441,6 +442,12 @@ public:
            //UNIMPLEMENTED;
 #endif
          }
+         if (m_warnings.length() > 0)
+         {
+             bDeleteFile = false;
+             throw MakeStringException(-1, "%s", m_warnings.str());
+         }
+
          break;
        }
        catch (IException* e)
@@ -465,7 +472,7 @@ public:
          e->Release();
 
          //remove incomplete (invalid) output file produced thus far
-         if (!DeleteFile(target))
+         if (bDeleteFile && !DeleteFile(target))
            WARNLOG("Couldn't delete file %s", target);
        }
        catch (...)

deployment/deploy/deploy.cpp

@@ -267,24 +267,7 @@ public:
             m_transform->transform( outputXml );
             m_transform->closeResultTarget();
             
-            const char* msg = m_transform->getMessages();
-            if (msg && *msg)
-            {
-                /*
-                //there may be multiple warnings messages bundled here so process each of them:
-                StringArray msgs;
-                DelimToStringArray(msg, msgs, "\n");
-                
-                ForEachItemIn(idx, msgs)
-                {
-                    msg = msgs.item(idx);
-                    if (msg && *msg)
-                    m_pCallback->printStatus(STATUS_NORMAL, NULL, NULL, NULL, msg);
-                }
-                */
-                m_sValidationErrors.append(msg);
-                m_nValidationErrors++;
-            }
+            m_pCallback->printStatus(STATUS_NORMAL, NULL, NULL, NULL, m_transform->getMessages());
 
             if (!m_nValidationErrors)//this may get filled in by the external function
                 valid = true;

initfiles/componentfiles/thor/run_thor

@@ -54,6 +54,14 @@ while [ 1 ]; do
     else
         nslaves=`cat $instancedir/uslaves.start | wc -l`
         $deploydir/frunssh $instancedir/uslaves.start "/bin/sh -c '$deploydir/start_slaves %a $THORMASTER $THORMASTERPORT $logdir $instancedir $deploydir $THORNAME $PATH_PRE $logredirect'" -i:$SSHidentityfile -u:$SSHusername -pe:$SSHpassword -t:$SSHtimeout -a:$SSHretries -n:$nslaves 2>&1
+        FRUNSSH_RC=$?
+        if [[ ${FRUNSSH_RC} -gt 0 ]]; then
+          echo "Error ${FRUNSSH_RC} in frunssh"
+          echo "Please check `dirname ${logdir}`/frunssh for more details"
+          # clean up any slaves it was able to reach
+          $deploydir/stop_thor $deploydir
+          exit 0
+        fi
     fi
 
     echo thormaster cmd : $instancedir/thormaster_$THORNAME MASTER=$THORMASTER:$THORMASTERPORT

initfiles/componentfiles/thor/stop_slaves

@@ -20,9 +20,6 @@ hpcc_setenv=$2
 
 source ${hpcc_setenv}
 
-echo compName=${compName}
-echo PID=${PID}
-
 slavename=thorslave_${compName}
 
 killall -0 $slavename &> /dev/null

plugins/javaembed/javaembed.cpp

@@ -121,9 +121,13 @@ public:
             optionStrings.append(libPath);
         }
 
+        // Options we should set (but allow for override with jvmoptions below)
+        optionStrings.append("-XX:-UseLargePages");
+
         if (conf && conf->hasProp("jvmoptions"))
         {
-            optionStrings.appendList(conf->queryProp("jvmoptions"), ENVSEPSTR);
+            // Use space as field sep as ':' and ';' are valid
+            optionStrings.appendList(conf->queryProp("jvmoptions"), " ");
         }
 
         // Options we know we always want set

plugins/redis/README.md

@@ -17,6 +17,12 @@ The redis server and client software can be obtained via either - [binaries](htt
 sudo apt-get redis-server
 ```
 
+*Note:* redis-server 2.6.12 or greater is required to use this plugin as intended. For efficiency, such version requirments are not checked as this is a runtime check only. The use of a
+lesser version will result in an exception, normally indicating that either a given command does not exist or that the wrong number of arguments was passed to it. The Set<type>
+plugin functions will not work when setting with an expiration for a version less than 2.6.12. In addition, whilst it is possible to use `Expire` with a version less than
+2.1.3 it is not advised due to [the change in its semantics](http://redis.io/commands/expire).
+
+
 Getting started
 ---------------
 
@@ -30,7 +36,7 @@ on different ports.
 The **redis-server** package comes with the redis client **redis-cli**. This can be used to send and receive commands to and from the server, invoked by `redis-cli` or, for example,
 `redis-cli -p 6380` to connect to the redis-cache on port 6380 (assuming one has been started).
 
-Perhaps on of the most handy uses of **redis-cli** is the ability to monitor all commands issued to the server via the redis command `MONITOR`. `INFO ALL` is also a useful command
+Perhaps one of the most handy uses of **redis-cli** is the ability to monitor all commands issued to the server via the redis command `MONITOR`. `INFO ALL` is also a useful command
 for listing the server and cache settings and statistics. *Note:* that if **requirepass** is activated **redis-cli** with require you to authenticate via `AUTH <passcode>`.
 
 Further [documentation](http://redis.io/documentation) is available with a full list of redis [commands](http://redis.io/commands).
@@ -38,10 +44,10 @@ Further [documentation](http://redis.io/documentation) is available with a full
 The Actual Plugin
 -----------------
 
-The bulk of this redis plugin for **ECL** is made up of the various `SET` and `GET` commands e.g. `GetString` or `SetReal`. They are accessible via the module `sync`
+The bulk of this redis plugin for **ECL** is made up of the various `SET` and `GET` commands e.g. `GetString` or `SetReal`. They are accessible via the module `redis`
 from the redis plugin **ECL** library `lib-redis`. i.e.
 ```
-IMPORT sync FROM lib_redis;
+IMPORT redis FROM lib_redis;
 ```
 Here is a list of the core plugin **functions**.
 
@@ -90,7 +96,7 @@ The core points to note here are:
    * `UNSIGNED expire` has a default of **0**, i.e. *forever*.
 
 ###The redisServer MODULE
-To avoid the combersom and unnecessary need to constantly pass `options` and `password` with each function call, the module `redisServer` can be imported to effectively 
+To avoid the cumbersome and unnecessary need to constantly pass `options` and `password` with each function call, the module `redisServer` can be imported to effectively 
 *wrap* the above functions.
 ```
 IMPORT redisServer FROM lib_redis;
@@ -116,9 +122,9 @@ Race Retrieval and Locking Keys
 A common use of external caching systems such as **redis** is for temporarily storing data that may be expensive, computationally or otherwise, to obtain and thus doing so
 *only once* is paramount. In such a scenario it is possible (in cases usual) for multiple clients/requests to *hit* the cache simultaneously and upon finding that the data
 requested has not yet been stored, it is desired that only one of such requests obtain the new value and then store it for the others to then also obtain (from the cache).
-This plugin offers a solution to such a problem via the `GetOrLock` and `SetAndPublish` functions within the `redisServer` and `sync` modules of lib_redis.
+This plugin offers a solution to such a problem via the `GetOrLock` and `SetAndPublish` functions within the `redisServer` and `redis` modules of lib_redis.
 This module contains only three function categories - the `SET` and `GET` functions for **STRING**, **UTF8**, and **UNICODE** (i.e. only those that return empty strings)
-and lastley, an auxiliary function `Unlock` used to manually unlock locked keys as it be discussed.
+and lastly, an auxiliary function `Unlock` used to manually unlock locked keys as it be discussed.
 
 The principle here is based around a *cache miss* in which a requested key does not exist, the first requester (*race winner*) 'locks' the key in an atomic fashion.
 Any other simultaneous requester (*race loser*) finds that the key exists but has been locked and thus **SUBSCRIBES** to the key awaiting a **PUBLICATION** message
@@ -159,3 +165,24 @@ A few notes to point out here:
     have the key-value received on another, and yet, the key-value still does not exist on the cache.
    * At present the 'lock' is not as such an actual lock, as only the `locking.Get<type>` functions acknowledge it. By current implementation it is better thought as a flag for
    `GET` to wait and subscribe. I.e. the locked key can be deleted and re-set just as any other key can be.
+   * Since the timeout duration is not for an individual plugin call but instead that waiting for each reply from the server, the actual possible maximum timeout duration differs from
+     various functions within this plugin, i.e. some functions do more than others. Below is a table for each of the plugin functions (or catagories of) including the maximum possible and
+     nominal expected, where the latter is due to using a cached connection, i.e. neither the server IP, port, nor password have changed from the function called prior to the one in
+     question. The values given are multiples of the given timeout.
+
+| Operation/Function  | Nominal | Maximum | Diff due to...   |
+|:--------------------|:-------:|:-------:|-----------------:|
+| A new connection    | 3       | 4       | database         |
+| Cached connection   | 0       | 2       | database, timeout|
+| Get<type>           | 1       | 5       | new connection   |
+| Set<type>           | 1       | 5       | new connection   |
+| FlushDB             | 1       | 5       | new connection   |
+| Del                 | 1       | 5       | new connection   |
+| Persist             | 1       | 5       | new connection   |
+| Exists              | 1       | 5       | new connection   |
+| DBSize              | 1       | 5       | new connection   |
+| Expire              | 1       | 5       | new connection   |
+| GetOrLock           | 7       | 11      | new connection   |
+| GetOrLock (locked)  | 8       | 12      | new connection   |
+| SetAndPublish       | 2       | 6       | new connection   |
+| Unlock              | 5       | 9       | new connection   |

plugins/redis/lib_redis.ecllib

@@ -16,7 +16,7 @@
 ############################################################################## */
 
 
-EXPORT sync := SERVICE : plugin('redis'), namespace('RedisPlugin')
+EXPORT redis := SERVICE : plugin('redis'), namespace('RedisPlugin')
   SetUnicode( CONST VARSTRING key, CONST UNICODE value, CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUChar';
   SetString(  CONST VARSTRING key, CONST STRING value,  CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetStr';
   SetUtf8(    CONST VARSTRING key, CONST UTF8 value,    CONST VARSTRING options, UNSIGNED database = 0, UNSIGNED4 expire = 0, CONST VARSTRING password = '', UNSIGNED timeout = 1000) : cpp,action,context,entrypoint='SyncRSetUtf8';
@@ -55,39 +55,39 @@ EXPORT sync := SERVICE : plugin('redis'), namespace('RedisPlugin')
 END;
 
 EXPORT RedisServer(VARSTRING options, VARSTRING password = '', UNSIGNED timeout = 1000) := MODULE
-  EXPORT  SetUnicode(VARSTRING key, UNICODE value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetUnicode (key, value, options, database, expire, password, timeout);
-  EXPORT   SetString(VARSTRING key, STRING value,   UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetString  (key, value, options, database, expire, password, timeout);
-  EXPORT     SetUtf8(VARSTRING key, UTF8 value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetUtf8    (key, value, options, database, expire, password, timeout);
-  EXPORT  SetBoolean(VARSTRING key, BOOLEAN value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetBoolean (key, value, options, database, expire, password, timeout);
-  EXPORT     SetReal(VARSTRING key, REAL value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetReal    (key, value, options, database, expire, password, timeout);
-  EXPORT  SetInteger(VARSTRING key, INTEGER value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetInteger (key, value, options, database, expire, password, timeout);
-  EXPORT SetUnsigned(VARSTRING key, UNSIGNED value, UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetUnsigned(key, value, options, database, expire, password, timeout);
-  EXPORT     SetData(VARSTRING key, DATA value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetData    (key, value, options, database, expire, password, timeout);
-
-  EXPORT  GetUnicode(VARSTRING key, UNSIGNED database = 0) :=  sync.GetUnicode(key, options, database, password, timeout);
-  EXPORT   GetString(VARSTRING key, UNSIGNED database = 0) :=   sync.GetString(key, options, database, password, timeout);
-  EXPORT     GetUtf8(VARSTRING key, UNSIGNED database = 0) :=     sync.GetUtf8(key, options, database, password, timeout);
-  EXPORT  GetBoolean(VARSTRING key, UNSIGNED database = 0) :=  sync.GetBoolean(key, options, database, password, timeout);
-  EXPORT     GetReal(VARSTRING key, UNSIGNED database = 0) :=     sync.GetReal(key, options, database, password, timeout);
-  EXPORT  GetInteger(VARSTRING key, UNSIGNED database = 0) :=  sync.GetInteger(key, options, database, password, timeout);
-  EXPORT GetUnsigned(VARSTRING key, UNSIGNED database = 0) := sync.GetUnsigned(key, options, database, password, timeout);
-  EXPORT     GetData(VARSTRING key, UNSIGNED database = 0) :=     sync.GetData(key, options, database, password, timeout);
-
-  EXPORT Exists(VARSTRING key, UNSIGNED database = 0) := sync.Exists(key, options, database, password, timeout);
-  EXPORT FlushDB(UNSIGNED database = 0) := sync.FlushDB(options, database, password, timeout);
-  EXPORT Del(VARSTRING key, UNSIGNED database = 0) := sync.Del(key, options, database, password, timeout);
-  EXPORT Delete(VARSTRING key, UNSIGNED database = 0) := sync.Delete(key, options, database, password, timeout);
-  EXPORT Persist(VARSTRING key, UNSIGNED database = 0) := sync.Persist(key, options, database, password, timeout);
-  EXPORT Expire(VARSTRING key, UNSIGNED database = 0, UNSIGNED4 expire)  := sync.Expire(key, options, database, expire, password, timeout);
-  EXPORT DBSize(UNSIGNED database = 0) := sync.DBSize(options, database, password, timeout);
-
-  EXPORT  SetAndPublishUnicode(VARSTRING key, UNICODE value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetAndPublishUnicode (key, value, options, database, expire, password, timeout);
-  EXPORT   SetAndPublishString(VARSTRING key, STRING value,   UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetAndPublishString  (key, value, options, database, expire, password, timeout);
-  EXPORT     SetAndPublishUtf8(VARSTRING key, UTF8 value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := sync.SetAndPublishUtf8    (key, value, options, database, expire, password, timeout);
-
-  EXPORT  GetOrLockUnicode(VARSTRING key, UNSIGNED database = 0) :=  sync.GetOrLockUnicode(key, options, database, password, timeout);
-  EXPORT   GetOrLockString(VARSTRING key, UNSIGNED database = 0) :=   sync.GetOrLockString(key, options, database, password, timeout);
-  EXPORT     GetOrLockUtf8(VARSTRING key, UNSIGNED database = 0) :=     sync.GetOrLockUtf8(key, options, database, password, timeout);
-
-  EXPORT Unlock(VARSTRING key, UNSIGNED database = 0) := sync.Unlock(key, options, database, password, timeout);
+  EXPORT  SetUnicode(VARSTRING key, UNICODE value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetUnicode (key, value, options, database, expire, password, timeout);
+  EXPORT   SetString(VARSTRING key, STRING value,   UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetString  (key, value, options, database, expire, password, timeout);
+  EXPORT     SetUtf8(VARSTRING key, UTF8 value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetUtf8    (key, value, options, database, expire, password, timeout);
+  EXPORT  SetBoolean(VARSTRING key, BOOLEAN value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetBoolean (key, value, options, database, expire, password, timeout);
+  EXPORT     SetReal(VARSTRING key, REAL value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetReal    (key, value, options, database, expire, password, timeout);
+  EXPORT  SetInteger(VARSTRING key, INTEGER value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetInteger (key, value, options, database, expire, password, timeout);
+  EXPORT SetUnsigned(VARSTRING key, UNSIGNED value, UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetUnsigned(key, value, options, database, expire, password, timeout);
+  EXPORT     SetData(VARSTRING key, DATA value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetData    (key, value, options, database, expire, password, timeout);
+
+  EXPORT  GetUnicode(VARSTRING key, UNSIGNED database = 0) :=  redis.GetUnicode(key, options, database, password, timeout);
+  EXPORT   GetString(VARSTRING key, UNSIGNED database = 0) :=   redis.GetString(key, options, database, password, timeout);
+  EXPORT     GetUtf8(VARSTRING key, UNSIGNED database = 0) :=     redis.GetUtf8(key, options, database, password, timeout);
+  EXPORT  GetBoolean(VARSTRING key, UNSIGNED database = 0) :=  redis.GetBoolean(key, options, database, password, timeout);
+  EXPORT     GetReal(VARSTRING key, UNSIGNED database = 0) :=     redis.GetReal(key, options, database, password, timeout);
+  EXPORT  GetInteger(VARSTRING key, UNSIGNED database = 0) :=  redis.GetInteger(key, options, database, password, timeout);
+  EXPORT GetUnsigned(VARSTRING key, UNSIGNED database = 0) := redis.GetUnsigned(key, options, database, password, timeout);
+  EXPORT     GetData(VARSTRING key, UNSIGNED database = 0) :=     redis.GetData(key, options, database, password, timeout);
+
+  EXPORT Exists(VARSTRING key, UNSIGNED database = 0) := redis.Exists(key, options, database, password, timeout);
+  EXPORT FlushDB(UNSIGNED database = 0) := redis.FlushDB(options, database, password, timeout);
+  EXPORT Del(VARSTRING key, UNSIGNED database = 0) := redis.Del(key, options, database, password, timeout);
+  EXPORT Delete(VARSTRING key, UNSIGNED database = 0) := redis.Delete(key, options, database, password, timeout);
+  EXPORT Persist(VARSTRING key, UNSIGNED database = 0) := redis.Persist(key, options, database, password, timeout);
+  EXPORT Expire(VARSTRING key, UNSIGNED database = 0, UNSIGNED4 expire)  := redis.Expire(key, options, database, expire, password, timeout);
+  EXPORT DBSize(UNSIGNED database = 0) := redis.DBSize(options, database, password, timeout);
+
+  EXPORT  SetAndPublishUnicode(VARSTRING key, UNICODE value,  UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishUnicode (key, value, options, database, expire, password, timeout);
+  EXPORT   SetAndPublishString(VARSTRING key, STRING value,   UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishString  (key, value, options, database, expire, password, timeout);
+  EXPORT     SetAndPublishUtf8(VARSTRING key, UTF8 value,     UNSIGNED database = 0, UNSIGNED4 expire = 0) := redis.SetAndPublishUtf8    (key, value, options, database, expire, password, timeout);
+
+  EXPORT  GetOrLockUnicode(VARSTRING key, UNSIGNED database = 0) :=  redis.GetOrLockUnicode(key, options, database, password, timeout);
+  EXPORT   GetOrLockString(VARSTRING key, UNSIGNED database = 0) :=   redis.GetOrLockString(key, options, database, password, timeout);
+  EXPORT     GetOrLockUtf8(VARSTRING key, UNSIGNED database = 0) :=     redis.GetOrLockUtf8(key, options, database, password, timeout);
+
+  EXPORT Unlock(VARSTRING key, UNSIGNED database = 0) := redis.Unlock(key, options, database, password, timeout);
 END;

plugins/redis/redis.cpp

@@ -44,7 +44,11 @@ static const char * REDIS_LOCK_PREFIX = "redis_ecl_lock";
 static __thread Connection * cachedConnection;
 static __thread ThreadTermFunc threadHookChain;
 
-StringBuffer & appendExpire(StringBuffer & buffer, unsigned expire)
+static void * allocateAndCopy(const void * src, size_t size)
+{
+    return memcpy(rtlMalloc(size), src, size);
+}
+static StringBuffer & appendExpire(StringBuffer & buffer, unsigned expire)
 {
     if (expire > 0)
         buffer.append(" EX ").append(expire/1000);
@@ -112,20 +116,20 @@ protected :
     void parseOptions(ICodeContext * ctx, const char * _options);
     void connect(ICodeContext * ctx, unsigned __int64 _database, const char * password);
     void selectDB(ICodeContext * ctx, unsigned __int64 _database);
-    void authenticate(ICodeContext * ctx, const char * password);
     void resetContextErr();
     void readReply(Reply * reply);
     void readReplyAndAssert(Reply * reply, const char * msg);
     void readReplyAndAssertWithKey(Reply * reply, const char * msg, const char * key);
     void assertKey(const redisReply * reply, const char * key);
+    void assertAuthorization(const redisReply * reply);
     void assertOnError(const redisReply * reply, const char * _msg);
     void assertOnCommandError(const redisReply * reply, const char * cmd);
     void assertOnCommandErrorWithDatabase(const redisReply * reply, const char * cmd);
     void assertOnCommandErrorWithKey(const redisReply * reply, const char * cmd, const char * key);
     void assertConnection();
     void updateTimeout(unsigned __int64 _timeout);
-    void * allocateAndCopy(const char * src, size_t size);
-    bool isSameConnection(ICodeContext * ctx, const char * password) const;
+    static unsigned hashServerIpPortPassword(ICodeContext * ctx, const char * _options, const char * password);
+    bool isSameConnection(ICodeContext * ctx, const char * _options, const char * password) const;
 
     //-------------------------------LOCKING------------------------------------------------
     void handleLockOnSet(ICodeContext * ctx, const char * key, const char * value, size_t size, unsigned expire);
@@ -146,11 +150,11 @@ protected :
 
 //The following class is here to ensure destruction of the cachedConnection within the main thread
 //as this is not handled by the thread hook mechanism.
-static class mainThreadCachedConnection
+static class MainThreadCachedConnection
 {
 public :
-    mainThreadCachedConnection() { }
-    ~mainThreadCachedConnection()
+    MainThreadCachedConnection() { }
+    ~MainThreadCachedConnection()
     {
         if (cachedConnection)
             cachedConnection->Release();
@@ -171,10 +175,8 @@ static void releaseContext()
     }
 }
 Connection::Connection(ICodeContext * ctx, const char * _options, unsigned __int64 _database, const char * password, unsigned __int64 _timeout)
-  : database(0), timeout(_timeout), port(0)
+  : database(0), timeout(_timeout), port(0), serverIpPortPasswordHash(hashServerIpPortPassword(ctx, _options, password))
 {
-    serverIpPortPasswordHash = hashc((const unsigned char*)password, strlen(password), 0);
-    serverIpPortPasswordHash = hashc((const unsigned char*)_options, strlen(_options), serverIpPortPasswordHash);
     options.set(_options, strlen(_options));
     parseOptions(ctx, _options);
     connect(ctx, _database, password);
@@ -188,7 +190,7 @@ Connection::Connection(ICodeContext * ctx, const char * _options, const char * _
 }
 void Connection::connect(ICodeContext * ctx, unsigned __int64 _database, const char * password)
 {
-    struct timeval to = { timeout/1000, timeout%1000 };
+    struct timeval to = { timeout/1000, (timeout%1000)*1000 };
     context = redisConnectWithTimeout(ip.str(), port, to);
     redisSetTimeout(context, to);
     assertConnection();
@@ -196,11 +198,11 @@ void Connection::connect(ICodeContext * ctx, unsigned __int64 _database, const c
     //The following is the dissemination of the two methods authenticate(ctx, password) & selectDB(ctx, _database)
     //such that they may be pipelined to save an extra round trip to the server and back.
     if (password && *password)
-      redisAppendCommand(context, "AUTH %b", password, strlen(password));
+        redisAppendCommand(context, "AUTH %b", password, strlen(password));
 
     if (database != _database)
     {
-        VStringBuffer cmd("SELECT %" I64F "u", database);
+        VStringBuffer cmd("SELECT %" I64F "u", _database);
         redisAppendCommand(context, cmd.str());
     }
 
@@ -215,15 +217,13 @@ void Connection::connect(ICodeContext * ctx, unsigned __int64 _database, const c
         database = _database;
     }
 }
-bool Connection::isSameConnection(ICodeContext * ctx, const char * password) const
+bool Connection::isSameConnection(ICodeContext * ctx, const char * _options, const char * password) const
 {
-    unsigned hash = hashc((const unsigned char*)options.str(), options.length(), hashc((const unsigned char*)password, strlen(password), 0));
-    return (serverIpPortPasswordHash == hash);
+    return (hashServerIpPortPassword(ctx, _options, password) == serverIpPortPasswordHash);
 }
-void * Connection::allocateAndCopy(const char * src, size_t size)
+unsigned Connection::hashServerIpPortPassword(ICodeContext * ctx, const char * _options, const char * password)
 {
-    void * value = rtlMalloc(size);
-    return memcpy(value, src, size);
+    return hashc((const unsigned char*)_options, strlen(_options), hashc((const unsigned char*)password, strlen(password), 0));
 }
 void Connection::parseOptions(ICodeContext * ctx, const char * _options)
 {
@@ -259,15 +259,6 @@ void Connection::parseOptions(ICodeContext * ctx, const char * _options)
             ctx->logString(msg.str());
         }
     }
-    return;
-}
-void Connection::authenticate(ICodeContext * ctx, const char * password)
-{
-    if (password && *password)
-    {
-        OwnedReply reply = Reply::createReply(redisCommand(context, "AUTH %b", password, strlen(password)));
-        assertOnError(reply->query(), "server authentication failed");
-    }
 }
 void Connection::resetContextErr()
 {
@@ -302,7 +293,7 @@ Connection * Connection::createConnection(ICodeContext * ctx, const char * optio
         return LINK(cachedConnection);
     }
 
-    if (cachedConnection->isSameConnection(ctx, password))
+    if (cachedConnection->isSameConnection(ctx, options, password))
     {
         //MORE: should perhaps check that the connection has not expired (think hiredis REDIS_KEEPALIVE_INTERVAL is defaulted to 15s).
         //At present updateTimeout calls assertConnection.
@@ -313,6 +304,7 @@ Connection * Connection::createConnection(ICodeContext * ctx, const char * optio
     }
 
     cachedConnection->Release();
+    cachedConnection = NULL;
     cachedConnection = new Connection(ctx, options, _database, password, _timeout);
     return LINK(cachedConnection);
 }
@@ -331,7 +323,7 @@ void Connection::updateTimeout(unsigned __int64 _timeout)
         return;
     assertConnection();
     timeout = _timeout;
-    struct timeval to = { timeout/1000, timeout%1000 };
+    struct timeval to = { timeout/1000, (timeout%1000)*1000 };
     assertex(context);
     if (redisSetTimeout(context, to) != REDIS_OK)
     {
@@ -346,7 +338,7 @@ void Connection::updateTimeout(unsigned __int64 _timeout)
 }
 void Connection::assertOnError(const redisReply * reply, const char * _msg)
 {
-    if (!reply)//assertex(reply)?
+    if (!reply)//MORE: should this be assertex(reply) instead?
     {
         //There should always be a context error if no reply error
         assertConnection();
@@ -355,21 +347,14 @@ void Connection::assertOnError(const redisReply * reply, const char * _msg)
     }
     else if (reply->type == REDIS_REPLY_ERROR)
     {
-        if (strncmp(reply->str, "NOAUTH", 6) == 0)
-        {
-            VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
-            rtlFail(0, msg.str());
-        }
-        else
-        {
-            VStringBuffer msg("Redis Plugin: %s - %s", _msg, reply->str);
-            rtlFail(0, msg.str());
-        }
+        assertAuthorization(reply);
+        VStringBuffer msg("Redis Plugin: %s - %s", _msg, reply->str);
+        rtlFail(0, msg.str());
     }
 }
 void Connection::assertOnCommandErrorWithKey(const redisReply * reply, const char * cmd, const char * key)
 {
-    if (!reply)//assertex(reply)?
+    if (!reply)//MORE: should this be assertex(reply) instead?
     {
         //There should always be a context error if no reply error
         assertConnection();
@@ -378,16 +363,9 @@ void Connection::assertOnCommandErrorWithKey(const redisReply * reply, const cha
     }
     else if (reply->type == REDIS_REPLY_ERROR)
     {
-        if (strncmp(reply->str, "NOAUTH", 6) == 0)
-        {
-            VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
-            rtlFail(0, msg.str());
-        }
-        else
-        {
-            VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %" I64F "u failed : %s", cmd, key, database, reply->str);
-            rtlFail(0, msg.str());
-        }
+        assertAuthorization(reply);
+        VStringBuffer msg("Redis Plugin: ERROR - %s '%s' on database %" I64F "u failed : %s", cmd, key, database, reply->str);
+        rtlFail(0, msg.str());
     }
 }
 void Connection::assertOnCommandErrorWithDatabase(const redisReply * reply, const char * cmd)
@@ -401,16 +379,9 @@ void Connection::assertOnCommandErrorWithDatabase(const redisReply * reply, cons
     }
     else if (reply->type == REDIS_REPLY_ERROR)
     {
-        if (strncmp(reply->str, "NOAUTH", 6) == 0)
-        {
-            VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
-            rtlFail(0, msg.str());
-        }
-        else
-        {
-            VStringBuffer msg("Redis Plugin: ERROR - %s on database %" I64F "u failed : %s", cmd, database, reply->str);
-            rtlFail(0, msg.str());
-        }
+        assertAuthorization(reply);
+        VStringBuffer msg("Redis Plugin: ERROR - %s on database %" I64F "u failed : %s", cmd, database, reply->str);
+        rtlFail(0, msg.str());
     }
 }
 void Connection::assertOnCommandError(const redisReply * reply, const char * cmd)
@@ -424,16 +395,17 @@ void Connection::assertOnCommandError(const redisReply * reply, const char * cmd
     }
     else if (reply->type == REDIS_REPLY_ERROR)
     {
-        if (strncmp(reply->str, "NOAUTH", 6) == 0)
-        {
-            VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
-            rtlFail(0, msg.str());
-        }
-        else
-        {
-            VStringBuffer msg("Redis Plugin: ERROR - %s failed : %s", cmd, reply->str);
-            rtlFail(0, msg.str());
-        }
+        assertAuthorization(reply);
+        VStringBuffer msg("Redis Plugin: ERROR - %s failed : %s", cmd, reply->str);
+        rtlFail(0, msg.str());
+    }
+}
+void Connection::assertAuthorization(const redisReply * reply)
+{
+    if (strncmp(reply->str, "NOAUTH", 6) == 0)
+    {
+        VStringBuffer msg("Redis Plugin: server authentication failed - %s", reply->str);
+        rtlFail(0, msg.str());
     }
 }
 void Connection::assertKey(const redisReply * reply, const char * key)
@@ -706,7 +678,7 @@ void Connection::lockGet(ICodeContext * ctx, const char * key, size_t & returnSi
 //---------------------------------------------------------------------------------------
 void Connection::encodeChannel(StringBuffer & channel, const char * key) const
 {
-    channel.append(REDIS_LOCK_PREFIX).append("_").append(key).append("_").append(database).append("_").append(ip.str()).append("_").append(port);
+    channel.append(REDIS_LOCK_PREFIX).append("_").append(key).append("_").append(database);
 }
 bool Connection::lock(ICodeContext * ctx, const char * key, const char * channel)
 {
@@ -716,9 +688,7 @@ bool Connection::lock(ICodeContext * ctx, const char * key, const char * channel
     OwnedReply reply = Reply::createReply(redisCommand(context, cmd.str(), key, strlen(key), channel, strlen(channel)));
     assertOnError(reply->query(), cmd.append(" of the key '").append(key).append("' failed"));
 
-    if (reply->query()->type == REDIS_REPLY_STATUS && strcmp(reply->query()->str, "OK") == 0)
-        return true;
-    return false;
+    return (reply->query()->type == REDIS_REPLY_STATUS && strcmp(reply->query()->str, "OK") == 0);
 }
 void Connection::unlock(ICodeContext * ctx, const char * key)
 {
@@ -796,9 +766,9 @@ void Connection::handleLockOnGet(ICodeContext * ctx, const char * key, MemoryAtt
     else
     {
         //Check that we SUBSCRIBEd to the correct channel (which could have been manually SET).
-        if (strcmp(reply->query()->str, channel) !=0 )
+        if (strcmp(reply->query()->str, channel.str()) !=0 )
         {
-            VStringBuffer msg("Redis Plugin: ERROR - the key '%s', on database %" I64F "u, is locked with a channel ('%s') different to that subscribed to.", key, database, reply->query()->str);
+            VStringBuffer msg("Redis Plugin: ERROR - the key '%s', on database %" I64F "u, is locked with a channel ('%s') different to that subscribed to (%s).", key, database, reply->query()->str, channel.str());
             rtlFail(0, msg.str());
             //MORE: We could attempt to recover at this stage by subscribing to the channel that the key was actually locked with.
             //However, we may have missed the massage publication already or by then.
@@ -855,25 +825,25 @@ void Connection::handleLockOnSet(ICodeContext * ctx, const char * key, const cha
 //                           ECL SERVICE ENTRYPOINTS
 //--------------------------------------------------------------------------------
 //-----------------------------------SET------------------------------------------
-ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, size32_t valueSize, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
+ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetStr(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
 {
-    SyncLockRSet(ctx, options, key, valueSize, value, expire,  database, password, timeout);
-    returnSize = valueSize;
-    returnValue = (char*)memcpy(rtlMalloc(valueSize), value, valueSize);
+    SyncLockRSet(ctx, options, key, valueLength, value, expire,  database, password, timeout);
+    returnLength = valueLength;
+    returnValue = (char*)allocateAndCopy(value, valueLength);
 }
 ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUChar(ICodeContext * ctx, size32_t & returnLength, UChar * & returnValue, const char * key, size32_t valueLength, const UChar * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
 {
     unsigned valueSize = (valueLength)*sizeof(UChar);
     SyncLockRSet(ctx, options, key, valueSize, (char*)value, expire, database, password, timeout);
     returnLength= valueLength;
-    returnValue = (UChar*)memcpy(rtlMalloc(valueSize), (void*)value, valueSize);
+    returnValue = (UChar*)allocateAndCopy(value, valueSize);
 }
 ECL_REDIS_API void ECL_REDIS_CALL SyncLockRSetUtf8(ICodeContext * ctx, size32_t & returnLength, char * & returnValue, const char * key, size32_t valueLength, const char * value, const char * options, unsigned __int64 database, unsigned expire, const char * password, unsigned __int64 timeout)
 {
     unsigned valueSize = rtlUtf8Size(valueLength, value);
     SyncLockRSet(ctx, options, key, valueSize, value, expire, database, password, timeout);
     returnLength = valueLength;
-    returnValue = (char*)memcpy(rtlMalloc(valueSize), value, valueSize);
+    returnValue = (char*)allocateAndCopy(value, valueSize);
 }
 //-------------------------------------GET----------------------------------------
 ECL_REDIS_API void ECL_REDIS_CALL SyncLockRGetStr(ICodeContext * ctx, size32_t & returnSize, char * & returnValue, const char * key, const char * options, unsigned __int64 database, const char * password, unsigned __int64 timeout)

system/jlib/jutil.cpp

@@ -404,7 +404,7 @@ HINSTANCE LoadSharedObject(const char *name, bool isGlobal, bool raiseOnError)
         if (h == NULL)
         {
             StringBuffer dlErrorMsg(dlerror());
-            DBGLOG("Error loading %s: %s", name, dlErrorMsg.str());
+            DBGLOG("Warning: Could not load %s: %s", name, dlErrorMsg.str());
             if (raiseOnError)
             {
                 if (isCorruptDll(dlErrorMsg.str()))

system/xmllib/libxslt_processor.cpp

@@ -403,6 +403,7 @@ public:
         return exceptions.get();
     }
     void clearExceptions(){exceptions.clear();}
+    void clearMessages(){messages.clear();}
 
 public:
     Owned<IProperties> xslParameters;
@@ -480,6 +481,9 @@ static void libxsltErrorMsgHandler(void *ctx, const char *format, ...)
 {
     if (!ctx)
         return;
+    if (format && *format == '\n')
+        return;
+
     CLibXslTransform *ctrans = (CLibXslTransform*)ctx;//getXsltTransformObject((xsltTransformContextPtr)ctx);
     if (!ctrans)
         return;
@@ -492,6 +496,8 @@ static void libxsltErrorMsgHandler(void *ctx, const char *format, ...)
 int CLibXslTransform::transform(xmlChar **xmlbuff, int &len)
 {
     clearExceptions();
+    clearMessages();
+
     xsltSetGenericErrorFunc(this, libxsltErrorMsgHandler);
     if (!xmlSrc)
         throw MakeStringException(XSLERR_MissingXml, "XSLT Transform missing XML");
@@ -544,6 +550,8 @@ int CLibXslTransform::transform(xmlChar **xmlbuff, int &len)
         throw MakeStringException(XSLERR_TransformFailed, "Failed running xlst using libxslt.");
     }
 
+    xsltTransformState stateAfterTransform = ctxt->state;
+
     try
     {
         xsltFreeTransformContext(ctxt);
@@ -556,8 +564,20 @@ int CLibXslTransform::transform(xmlChar **xmlbuff, int &len)
         throw MakeStringException(XSLERR_TransformFailed, "Failed processing libxslt transform output");
     }
     xmlFreeDoc(res);
+
     if (exceptions && exceptions->ordinality())
-        throw exceptions.getClear();
+    {
+        if (stateAfterTransform != XSLT_STATE_OK)
+        {
+            throw exceptions.getClear();
+        }
+        else
+        {
+            StringBuffer strErrMsg;
+            exceptions.get()->errorMessage(strErrMsg);
+            messages.set(strErrMsg.str());
+        }
+    }
 
     return 0;
 }

testing/regress/ecl/join.ecl

@@ -66,74 +66,74 @@ recpair makePairUKSkip(Files.DG_FlatFile L, Files.DG_FlatFileEvens R, string nam
 Out19 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: simple inner'));
+      , makePairUK(left, right, 'Unkeyed: simple inner'), STREAMED);
 Out20 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: simple only'), LEFT ONLY);
+      , makePairUK(left, right, 'Unkeyed: simple only'), LEFT ONLY, STREAMED);
 Out21 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: simple outer'), LEFT OUTER);
+      , makePairUK(left, right, 'Unkeyed: simple outer'), LEFT OUTER, STREAMED);
 Out22 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUKSkip(left, right, 'Unkeyed: simple skip'));
+      , makePairUKSkip(left, right, 'Unkeyed: simple skip'), STREAMED);
 Out23 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUKSkip(left, right, 'Unkeyed: simple skip, left only'), LEFT ONLY);
+      , makePairUKSkip(left, right, 'Unkeyed: simple skip, left only'), LEFT ONLY, STREAMED);
 Out24 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUKSkip(left, right, 'Unkeyed: simple skip, left outer'), LEFT OUTER);
+      , makePairUKSkip(left, right, 'Unkeyed: simple skip, left outer'), LEFT OUTER, STREAMED);
 Out25 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
-      , makePairUK(left, right, 'Unkeyed: simple keep(2)'), KEEP(2));
+      , makePairUK(left, right, 'Unkeyed: simple keep(2)'), KEEP(2), STREAMED);
 Out26 :=JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
-      , makePairUK(left, right, 'Unkeyed: simple atmost(3)'), ATMOST(3));
+      , makePairUK(left, right, 'Unkeyed: simple atmost(3)'), ATMOST(3), STREAMED);
 Out27 :=choosen(JOIN(Files.DG_FlatFile, Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: simple choosen'), LEFT OUTER), 1);
+      , makePairUK(left, right, 'Unkeyed: simple choosen'), LEFT OUTER, STREAMED), 1);
 
 //unkeyedjoins(unkeyedgrouped, group(Files.DG_FlatFile, DG_firstname), 'grouped');
 //Out1 :=dataset([{'Unkeyed: grouped '}], {string80 __________________}));
 Out46 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: grouped inner'));
+      , makePairUK(left, right, 'Unkeyed: grouped inner'), STREAMED);
 Out47 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: grouped only'), LEFT ONLY);
+      , makePairUK(left, right, 'Unkeyed: grouped only'), LEFT ONLY, STREAMED);
 Out48 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: grouped outer'), LEFT OUTER);
+      , makePairUK(left, right, 'Unkeyed: grouped outer'), LEFT OUTER, STREAMED);
 Out49 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUKSkip(left, right, 'Unkeyed: grouped skip'));
+      , makePairUKSkip(left, right, 'Unkeyed: grouped skip'), STREAMED);
 Out50 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUKSkip(left, right, 'Unkeyed: grouped skip, left only'), LEFT ONLY);
+      , makePairUKSkip(left, right, 'Unkeyed: grouped skip, left only'), LEFT ONLY, STREAMED);
 Out51 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUKSkip(left, right, 'Unkeyed: grouped skip, left outer'), LEFT OUTER);
+      , makePairUKSkip(left, right, 'Unkeyed: grouped skip, left outer'), LEFT OUTER, STREAMED);
 Out52 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
-      , makePairUK(left, right, 'Unkeyed: grouped keep(2)'), KEEP(2));
+      , makePairUK(left, right, 'Unkeyed: grouped keep(2)'), KEEP(2), STREAMED);
 Out53 :=JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
-      , makePairUK(left, right, 'Unkeyed: grouped atmost(3)'), ATMOST(3));
+      , makePairUK(left, right, 'Unkeyed: grouped atmost(3)'), ATMOST(3), STREAMED);
 Out54 :=choosen(JOIN(group(Files.DG_FlatFile, DG_firstname), Files.DG_FlatFileEvens, left.DG_firstname = right.DG_firstname 
          AND left.DG_lastname=right.DG_lastname 
          AND left.DG_Prange=right.DG_Prange     
-      , makePairUK(left, right, 'Unkeyed: grouped choosen'), LEFT OUTER), 1);
+      , makePairUK(left, right, 'Unkeyed: grouped choosen'), LEFT OUTER, STREAMED), 1);
 
 output(SORT(Out19,record));
 output(SORT(Out20,record));

testing/regress/ecl/key/redislockingtest.xml

@@ -70,3 +70,9 @@
 <Dataset name='Result 24'>
  <Row><Result_24>false</Result_24></Row>
 </Dataset>
+<Dataset name='Result 25'>
+ <Row><value>Redis Plugin: ERROR - the key &apos;channelTest1&apos;, on database 0, is locked with a channel (&apos;redis_ecl_lock_blah_blah_blah&apos;) different to that subscribed to (redis_ecl_lock_channelTest1_0).</value></Row>
+</Dataset>
+<Dataset name='Result 26'>
+ <Row><value>RedisPlugin: ERROR - GET timed out.</value></Row>
+</Dataset>

testing/regress/ecl/key/redissynctest.xml

@@ -23,37 +23,37 @@
  <Row><Result_8>7</Result_8></Row>
 </Dataset>
 <Dataset name='Result 9'>
- <Row><Result_9>supercalifragilisticexpialidocious</Result_9></Row>
+ <Row><Result_9>7</Result_9></Row>
 </Dataset>
 <Dataset name='Result 10'>
- <Row><Result_10>false</Result_10></Row>
+ <Row><Result_10>supercalifragilisticexpialidocious</Result_10></Row>
 </Dataset>
 <Dataset name='Result 11'>
- <Row><Result_11>אבגדהוזחטיךכלםמןנסעףפץצקרשת</Result_11></Row>
+ <Row><Result_11>false</Result_11></Row>
 </Dataset>
 <Dataset name='Result 12'>
  <Row><Result_12>אבגדהוזחטיךכלםמןנסעףפץצקרשת</Result_12></Row>
 </Dataset>
 <Dataset name='Result 13'>
- <Row><Result_13>true</Result_13></Row>
+ <Row><Result_13>אבגדהוזחטיךכלםמןנסעףפץצקרשת</Result_13></Row>
 </Dataset>
 <Dataset name='Result 14'>
- <Row><Result_14>false</Result_14></Row>
+ <Row><Result_14>true</Result_14></Row>
 </Dataset>
 <Dataset name='Result 15'>
- <Row><Result_15>D790D791D792D793D794D795D796D798D799D79AD79BD79CD79DD79DD79ED79FD7A0D7A1D7A2D7A3D7A4D7A5D7A6D7A7D7A8D7A9D7AA</Result_15></Row>
+ <Row><Result_15>false</Result_15></Row>
 </Dataset>
 <Dataset name='Result 16'>
- <Row><Result_16>7523094288207667809</Result_16></Row>
+ <Row><Result_16>D790D791D792D793D794D795D796D798D799D79AD79BD79CD79DD79DD79ED79FD7A0D7A1D7A2D7A3D7A4D7A5D7A6D7A7D7A8D7A9D7AA</Result_16></Row>
 </Dataset>
 <Dataset name='Result 17'>
- <Row><Result_17>false</Result_17></Row>
+ <Row><Result_17>7523094288207667809</Result_17></Row>
 </Dataset>
 <Dataset name='Result 18'>
  <Row><Result_18>false</Result_18></Row>
 </Dataset>
 <Dataset name='Result 19'>
- <Row><Result_19>true</Result_19></Row>
+ <Row><Result_19>false</Result_19></Row>
 </Dataset>
 <Dataset name='Result 20'>
  <Row><Result_20>true</Result_20></Row>
@@ -62,26 +62,35 @@
  <Row><Result_21>true</Result_21></Row>
 </Dataset>
 <Dataset name='Result 22'>
- <Row><Result_22>false</Result_22></Row>
+ <Row><Result_22>true</Result_22></Row>
 </Dataset>
 <Dataset name='Result 23'>
- <Row><Result_23>Woof</Result_23></Row>
+ <Row><Result_23>false</Result_23></Row>
 </Dataset>
 <Dataset name='Result 24'>
- <Row><Result_24>Grrrr</Result_24></Row>
+ <Row><Result_24>Woof</Result_24></Row>
 </Dataset>
 <Dataset name='Result 25'>
- <Row><Result_25>Woof-Woof</Result_25></Row>
+ <Row><Result_25>Grrrr</Result_25></Row>
 </Dataset>
 <Dataset name='Result 26'>
- <Row><Result_26>5</Result_26></Row>
+ <Row><Result_26>Woof-Woof</Result_26></Row>
 </Dataset>
 <Dataset name='Result 27'>
- <Row><Result_27>2</Result_27></Row>
+ <Row><Result_27>5</Result_27></Row>
 </Dataset>
 <Dataset name='Result 28'>
- <Row><Result_28>0</Result_28></Row>
+ <Row><Result_28>2</Result_28></Row>
 </Dataset>
 <Dataset name='Result 29'>
- <Row><Result_29>300</Result_29></Row>
+ <Row><Result_29>0</Result_29></Row>
+</Dataset>
+<Dataset name='Result 30'>
+ <Row><Result_30>300</Result_30></Row>
+</Dataset>
+<Dataset name='Result 31'>
+ <Row><value>Redis Plugin: server authentication failed - NOAUTH Authentication required.</value></Row>
+</Dataset>
+<Dataset name='Result 32'>
+ <Row><value>Redis Plugin: ERROR - the requested key &apos;authTest1&apos; does not exist on database 0</value></Row>
 </Dataset>

testing/regress/ecl/redislockingtest.ecl

@@ -120,3 +120,18 @@ SEQUENTIAL(
     myRedis.Exists('testlock'),
     myRedis.FlushDB(),
     );
+
+//Test exception for checking expected channels
+ds1 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis.GetOrLockString('channelTest' + (string)COUNTER)));
+SEQUENTIAL(
+    myRedis.FlushDB();
+    myRedis.SetString('channelTest1', 'redis_ecl_lock_blah_blah_blah');
+    OUTPUT(CATCH(ds1, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );
+
+ds2 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis.GetOrLockString('channelTest' + (string)(1+COUNTER))));
+SEQUENTIAL(
+    myRedis.FlushDB();
+    myRedis.SetString('channelTest2', 'redis_ecl_lock_channelTest2_0');
+    OUTPUT(CATCH(ds2, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );

testing/regress/ecl/redissynctest.ecl

@@ -15,16 +15,16 @@
     limitations under the License.
 ############################################################################## */
 
-IMPORT sync FROM lib_redis;
+IMPORT redis FROM lib_redis;
 IMPORT Std;
 
 STRING server := '--SERVER=127.0.0.1:6379';
 STRING password := 'foobared';
-sync.FlushDB(server, /*database*/, password);
+redis.FlushDB(server, /*database*/, password);
 
 SEQUENTIAL(
-    sync.SetBoolean('b', TRUE, server, /*database*/, /*expire*/, password);
-    sync.GetBoolean('b', server, /*database*/, password);
+    redis.SetBoolean('b', TRUE, server, /*database*/, /*expire*/, password);
+    redis.GetBoolean('b', server, /*database*/, password);
     );
 
 IMPORT redisServer FROM lib_redis;
@@ -66,6 +66,12 @@ SEQUENTIAL(
     myRedis.GetUnsigned('u');
     );
 
+myRedis3 := RedisServer('--SERVER=127.0.0.1:6381', password);
+SEQUENTIAL(
+    myRedis3.SetUnsigned('u3', u);
+    myRedis3.GetUnsigned('u3');
+    );
+
 STRING str  := 'supercalifragilisticexpialidocious';
 SEQUENTIAL(
     myRedis.SetString('str', str);
@@ -147,7 +153,21 @@ s2 := DATASET(N, TRANSFORM({ integer a }, SELF.a := myRedis.GetInteger('transfor
 SEQUENTIAL(
     myRedis.SetInteger('transformTest', x),
     OUTPUT(SUM(NOFOLD(s1 + s2), a))//answer = (x+x/2)*N, in this case 300.
-);
+    );
 
+//Test some authentication exceptions
+myRedis4 := RedisServer(server);
+ds1 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis4.GetString('authTest' + (string)COUNTER)));
+SEQUENTIAL(
+    myRedis.FlushDB();
+    OUTPUT(CATCH(ds1, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );
+
+ds2 := DATASET(NOFOLD(1), TRANSFORM({string value}, SELF.value := myRedis.GetString('authTest' + (string)COUNTER)));
+SEQUENTIAL(
+    myRedis.FlushDB();
+    OUTPUT(CATCH(ds2, ONFAIL(TRANSFORM({ STRING value }, SELF.value := FAILMESSAGE))));
+    );
+    
 myRedis.FlushDB();
 myRedis2.FlushDB();

thorlcr/activities/thactivityutil.cpp

@@ -782,6 +782,8 @@ IFileIO *createMultipleWrite(CActivityBase *activity, IPartDescriptor &partDesc,
     Owned<IFileIO> fileio;
     if (compress)
     {
+        if (activity->getOptBool(THOROPT_COMP_FORCELZW, false))
+            recordSize = 0; // by default if fixed length (recordSize set), row diff compression is used. This forces LZW
         fileio.setown(createCompressedFileWriter(file, recordSize, extend, true, ecomp));
         if (!fileio)
         {
@@ -795,7 +797,7 @@ IFileIO *createMultipleWrite(CActivityBase *activity, IPartDescriptor &partDesc,
         fileio.setown(file->open(extend&&file->exists()?IFOwrite:IFOcreate)); 
     if (!fileio)
         throw MakeActivityException(activity, TE_FileCreationFailed, "Failed to create file for write (%s) error = %d", outLocationName.str(), GetLastError());
-    ActPrintLog(activity, "Writing to file: %s", file->queryFilename());
+    ActPrintLog(activity, "Writing to file: %s, compress=%s, rdiff=%s", file->queryFilename(), compress ? "true" : "false", (compress && recordSize) ? "true" : "false");
     return new CWriteHandler(*activity, partDesc, file, fileio, iProgress, direct, renameToPrimary, aborted);
 }
 

thorlcr/activities/thdiskbaseslave.cpp

@@ -314,7 +314,16 @@ void CDiskWriteSlaveActivityBase::open()
     bool extend = 0 != (diskHelperBase->getFlags() & TDWextend);
     if (extend)
         ActPrintLog("Extending file %s", fName.get());
-    size32_t exclsz = 0;
+
+    size32_t diskRowMinSz = 0;
+    IOutputMetaData *diskRowMeta = diskHelperBase->queryDiskRecordSize()->querySerializedDiskMeta();
+    if (diskRowMeta->isFixedSize())
+    {
+        diskRowMinSz = diskRowMeta->getMinRecordSize();
+        if (grouped)
+            diskRowMinSz += 1;
+    }
+
     calcFileCrc = true;
 
     bool external = dlfn.isExternal();
@@ -324,7 +333,7 @@ void CDiskWriteSlaveActivityBase::open()
 
     bool direct = query || (external && !firstNode());
     bool rename = !external || (!query && lastNode());
-    Owned<IFileIO> iFileIO = createMultipleWrite(this, *partDesc, exclsz, compress, extend||(external&&!query), ecomp, this, direct, rename, &abortSoon, (external&&!query) ? &tempExternalName : NULL);
+    Owned<IFileIO> iFileIO = createMultipleWrite(this, *partDesc, diskRowMinSz, compress, extend||(external&&!query), ecomp, this, direct, rename, &abortSoon, (external&&!query) ? &tempExternalName : NULL);
 
     if (compress)
     {

thorlcr/thorutil/thormisc.hpp

@@ -68,6 +68,7 @@
 #define THOROPT_LKJOIN_LOCALFAILOVER  "lkjoin_localfailover"    // Force SMART to failover to distributed local lookup join (for testing only)   (default = false)
 #define THOROPT_LKJOIN_HASHJOINFAILOVER "lkjoin_hashjoinfailover" // Force SMART to failover to hash join (for testing only)                     (default = false)
 #define THOROPT_MAX_KERNLOG           "max_kern_level"          // Max kernel logging level, to push to workunit, -1 to disable                  (default = 3)
+#define THOROPT_COMP_FORCELZW         "forceLZW"                // Forces file compression to use LZW                                            (default = false)
 
 #define INITIAL_SELFJOIN_MATCH_WARNING_LEVEL 20000  // max of row matches before selfjoin emits warning