How to release unused XiNodes from BW process variables

BW accumulates variables across job lifetime. Very often intermediate variables in later steps are not needed any more, but still occupy memory. It is possible to clean them and make more memory for new jobs, globally reducing RAM requirement for whole BW instance.

package Libraries.Utils.ReleaseVariable;
import java.lang.reflect.*;
import java.util.*;
import java.util.concurrent.atomic.*;
import java.io.*;
import org.xml.sax.InputSource;

import com.tibco.xml.datamodel.XiNode;
import com.tibco.xml.datamodel.XiParserFactory;
import com.tibco.xml.xdata.xpath.Variable;
import com.tibco.xml.xdata.xpath.VariableList;
import com.tibco.pe.core.*;

public class ReleaseVariableJavaCode{

 public static JobPool getJobPool() {
  for (Field f : Engine.class.getDeclaredFields()) {
   if (f.getType().getName().endsWith("JobPool")) {
    f.setAccessible(true);
    try {
     return (JobPool) f.get(null);
    }
    catch (Throwable te) {
     throw new RuntimeException("Cannot access JobPool: "+te.getMessage(), te);
    }    
   }   
  }
  throw new RuntimeException("Cannot access JobPool");
 }

 public static long[] getJobIds() {
  try {
   return getJobPool().getJobIds();
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access JobPool.getJobIds(): "+t.getMessage(), t);
  }
 }
 
 public static VariableList getJobVariables(long jid, int trackId) {
  try {
   Method getAttributes = Class.forName("com.tibco.pe.core.Job").getDeclaredMethod("getAttributes", new Class<?>[] { int.class });
   getAttributes.setAccessible(true);
   return (VariableList) getAttributes.invoke(getJobPool().findJob(jid), trackId);
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access job variables: "+t.getMessage(), t);
  }
 }

 public int getTrackId(long jid) {
  try {
   Method getter = Class.forName("com.tibco.pe.core.Job").getDeclaredMethod("getTrackId", new Class<?>[0]);
   getter.setAccessible(true);
   return (int) getter.invoke(getJobPool().findJob(jid));
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access job methods: "+t.getMessage(), t);
  }
 }

 public void nullifyXiNode(XiNode node) {
  while (node!=null && node.hasChildNodes())
   node.removeChild(node.getLastChild());
  if (callGC==1)
   System.gc();
 }

 public void prune(XiNode node, String[] tokens, int i) {
  //System.out.println("prune at level "+i+" of "+node);
  if (i==1)
   node = node.hasChildNodes() ? node.getFirstChild() : null; // root node
  while (node!=null) {
   //System.out.println("node name is "+(node.getName()!=null ? node.getName().getLocalName() : null));
   if (node.getName() != null && tokens[i].equals(node.getName().getLocalName())) {
    if (i == tokens.length-1) {
     //System.out.println("nullify");
     nullifyXiNode(  node );
    }
    else {
     XiNode nd = node.hasChildNodes() ? node.getFirstChild() : null;
     if (nd!=null) {
      prune(nd, tokens, i+1);      
     }
    }
   }
   node = node.hasNextSibling() ? node.getNextSibling() : null;
  }
 }
/****** START SET/GET METHOD, DO NOT MODIFY *****/
 protected long jobId = 0;
 protected String var = "";
 protected int callGC = 0;
 public long getjobId() {
  return jobId;
 }
 public void setjobId(long val) {
  jobId = val;
 }
 public String getvar() {
  return var;
 }
 public void setvar(String val) {
  var = val;
 }
 public int getcallGC() {
  return callGC;
 }
 public void setcallGC(int val) {
  callGC = val;
 }
/****** END SET/GET METHOD, DO NOT MODIFY *****/
 public ReleaseVariableJavaCode() {
 }
 public void invoke() throws Exception {
/* Available Variables: DO NOT MODIFY
 In  : long jobId
 In  : String var
 In  : int callGC
* Available Variables: DO NOT MODIFY *****/
String[] tokens = var.split("/");
int currentTrackId = getTrackId(jobId);

for (int iter=0; iter <= 4; iter++) {
    VariableList varList = getJobVariables(jobId, iter);
    Variable v = varList!=null && iter != currentTrackId ? varList.getVariable(tokens[0]) : null;

    if (v!=null) { 
      if (tokens.length == 1) {
          nullifyXiNode(  v.getValue() ); 
      }
      else if (tokens.length>1) {
  prune( v.getValue(), tokens, 1); 
      }
    }
}


}
}

Aerospike sever has got synchronization bugs :(

Bugs found during VAT testing (http://1307723433353.blogspot.com/2014/05/vat-validation-of-architecture-in-tests.html)

Aerospike is Open Source and if you encounter a problem you can analyze it yourself without waiting for the support in a different timezone. You can recompile fixed code and bring it immediately to the production. It is a way faster than with any platinum level support. You need only one guy who knows Linux and C.

How to get memory usage of Tibco BW Job

package Libraries.Utils.GetJobsMemoryUsage;
import java.lang.reflect.*;
import java.util.*;
import java.util.concurrent.atomic.*;
import java.io.*;
import org.xml.sax.InputSource;

import com.tibco.xml.datamodel.XiNode;
import com.tibco.xml.datamodel.XiParserFactory;
import com.tibco.xml.xdata.xpath.Variable;
import com.tibco.xml.xdata.xpath.VariableList;
import com.tibco.pe.core.*;

public class GetJobsMemoryUsageJavaCode{

 public static JobPool getJobPool() {
  for (Field f : Engine.class.getDeclaredFields()) {
   if (f.getType().getName().endsWith("JobPool")) {
    f.setAccessible(true);
    try {
     return (JobPool) f.get(null);
    }
    catch (Throwable te) {
     throw new RuntimeException("Cannot access JobPool: "+te.getMessage(), te);
    }    
   }   
  }
  throw new RuntimeException("Cannot access JobPool");
 }

 public static long[] getJobIds() {
  try {
   return getJobPool().getJobIds();
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access JobPool.getJobIds(): "+t.getMessage(), t);
  }
 }

 public static String getJobName(long id) {
  try {
   Object job = getJobPool().findJob(id);
   if (job!=null) {
    Method getWorkflow = Class.forName("com.tibco.pe.core.Job").getDeclaredMethod("getWorkflow", new Class<?>[0]);
    getWorkflow.setAccessible(true);
    com.tibco.pe.core.Workflow w = (com.tibco.pe.core.Workflow)getWorkflow.invoke(job);
    return w!=null ? w.getName() : "no-workflow-name";
   }
   else
    return null;
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access Job data: "+t.getMessage(), t);
  }
 }

 public static VariableList getJobVariables(long jid, int trackId) {
  try {
   Method getAttributes = Class.forName("com.tibco.pe.core.Job").getDeclaredMethod("getAttributes", new Class<?>[] { int.class });
   getAttributes.setAccessible(true);
   return (VariableList) getAttributes.invoke(getJobPool().findJob(jid), trackId);
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access job variables: "+t.getMessage(), t);
  }
 }

 public int getTrackId(long jid) {
  try {
   Method getter = Class.forName("com.tibco.pe.core.Job").getDeclaredMethod("getTrackId", new Class<?>[0]);
   getter.setAccessible(true);
   return (int) getter.invoke(getJobPool().findJob(jid));
  }
  catch (Throwable t) {
   throw new RuntimeException("Cannot access job methods: "+t.getMessage(), t);
  }
 }
/****** START SET/GET METHOD, DO NOT MODIFY *****/
 protected long jobId = 0;
 protected String[] usage = null;
 public long getjobId() {
  return jobId;
 }
 public void setjobId(long val) {
  jobId = val;
 }
 public String[] getusage() {
  return usage;
 }
 public void setusage(String[] val) {
  usage = val;
 }
/****** END SET/GET METHOD, DO NOT MODIFY *****/
 public GetJobsMemoryUsageJavaCode() {
 }
 public void invoke() throws Exception {
/* Available Variables: DO NOT MODIFY
 In  : long jobId
 Out : String[] usage
* Available Variables: DO NOT MODIFY *****/

LinkedList<String> usageList = new LinkedList<String>();
long ids[] = jobId > 0 ? new long[] { jobId } : getJobIds();

if (ids!=null) {
 StringBuilder sb = new StringBuilder();
 for (long id : ids) {
  String jobName = getJobName(id);
  sb.setLength(0); 
  AtomicLong mem = new AtomicLong(0);
  int currentTrackId = getTrackId(id);

  for (int trackId = 0; trackId < 4; trackId++) {
   VariableList varList = trackId == currentTrackId ? null : getJobVariables(id, trackId);
   if (varList!=null) {  
    for (Object varName : varList.getVariableNames()) {
     Variable v = varList.getVariable(varName.toString());
     long size = 0;
     if (v.getValue()==null)
      size = (v.getNumber()+"").length();
     else {
      final AtomicInteger cnt = new AtomicInteger(0);
      OutputStream cos = new OutputStream() {
       public void write(int i) { cnt.addAndGet(4); }
       public void write(byte bytes[], int off, int len) throws IOException { cnt.addAndGet(len); }
      };
      com.tibco.xml.datamodel.helpers.XiSerializer.serialize( v.getValue(), cos, "utf-8", true );
      size = cnt.get();
     }
     sb.append(varName).append('[').append(trackId).append(']').append("=").append(size).append("|");  
     mem.addAndGet(size);
    } 
   }
  }  
  sb.insert(0, jobName+"="+mem.longValue()+"|");
  usageList.add(sb.toString());
 } 
}
usage = usageList.toArray(new String[0]);}
}

Aerospike goes AIO!

Modified source code is here. I need to give it back to be compliant with GNU AGPLv3 licence.

Horyzontalny transfer genów

http://io9.com/confirmation-that-photosynthesizing-sea-slugs-steal-gen-1683702602

Fix 'No space left on device' with btrfs

Add new disk. Then: btrfs device add /dev/sdd /. And it's fixed.

Aerospike or not Aerospike

[4] + clustering without the need for GFS2/RHEL cluster/NAS
[5] + out of box interDC replication
[3] + faster than O_SYNC file and DB
[2] + cheaper for IT operations than 'canonical solutions for JEE'
[2] + better support, because vendor 'does care'

[2] - new uknown product
[2] - needs extensive crash test/validation test
[4] - need to buy support
[2] - huge RAM usage
[4] - will the company remain on the market for a decade?

16:14

Aerospike - fast or safe?

Let's look at https://github.com/aerospike/aerospike-server/blob/180ed47a5fffc54b3e45faccb33c908bc189db2e/as/src/storage/drv_ssd.c and try to find open() system call and fsync() system call. We see that open flags like O_SYNC are parametrized. There is also loop for fsync with parametrized sleep time. Relevant configuration (https://github.com/aerospike/aerospike-server/blob/master/as/src/base/cfg.c) lives in enable-osync and fsync-max-sec. Now, check default values at http://www.aerospike.com/docs/reference/configuration/. We see that synchronous writes are disabled by default, and fsync is also disabled. This means that there are no safety guaranties for single Aerospike node (see also: flush-max-ms). Due to asd/OS/storage crash data will likely be lost or corrupted. Probabilistic situation for 3 nodes on different racks with battery backed SSD array is quite different and the combined risk is very small (if we do not consider whole datacenter crash) and therefore Aerospike is acceptable solution with standard sane data safety/integrity/availability criteria.

Nie tankuj na Orlenie

BMW i8

Godzina jazdy na torze tym cudem kosztuje 7 000 zł

Samo cudo kosztuje ponad 500 000 zł

Tibco EMS on Aerospike NoSQL cluster

It is possible to have EMS distributed, partitioned, fault tolerant and replicated without clustered filesystem? Just hook into system calls and override storage access. Aerospike NoSQL solution comes with C client. EMS is also written in C/C++. Everything is now possible. Sky is the limit! Prof of concept doesn't have compaction implemented. It can be achieved by as_query_where + delete.

Now some benchmarks. 20 sender threads send 50 000 1KB messages, in the same time 20 receivers is active. Time of piping 1 mln messages (1GB) is measured. EMS on BTRFS inside virtual OpenSUSE 13.2 needs 3651 seconds, while EMS on Aerospike NoSQL needs 1961 seconds. EMS on file is 86% slower than on Aerospike.

Update: raw device with O_SYNC time is 2754s.

JMS datastore replication: performance with enterprise server

source: sync-msgs.db 5GB, GFS2

target: EXT4

serial: WALL TIME: 121sec 133msec

sendfile: WALL TIME: 113sec 264msec

default: WALL TIME: 64sec 7msec

mmap: WALL TIME: 142sec 57msec

aio: WALL TIME: 8sec 791msec

Replace XML element content with regexp

replaceAll
<([a-zA-Z]+[0-9]*:)?([^\s<]*[eE]lement[^\s]*)([^>]*>)([^<]*)(<\/\1\2)
<$1$2$3fixedContent$5

Make more friendly Tibco BW cmdline Validator

Override class name in validateproject.tra

package com.tibco.ae.tools.designer.cmdline;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

import com.tibco.ae.designerapi.AEResource;
import com.tibco.ae.designerapi.DesignerError;
import com.tibco.ae.designerapi.DesignerFolder;
import com.tibco.ae.tools.designer.AEApplication;
import com.tibco.ae.tools.designer.AEDocument;
import com.tibco.ae.tools.designer.actions.ValidateProjectAction;
import com.tibco.ae.tools.vfileresourcestore.FileSystemVFileFactoryResourceStore;
import com.tibco.objectrepo.vfile.VFile;
import com.tibco.util.ListHashMap;
import com.tibco.util.ResourceManager;

public class ValidateBwProject extends DesignerCommandLineApp {

private TempAliases aliases = null;
private int errorCount = 0;
private int warningCount =0;

public void execute(String projectLocation, String aliasFileLocation) throws Exception {
initResources();
initializeApp();
TempAliases temp = new TempAliases() {
public void updateAliases(File file) throws FileNotFoundException, IOException {
int start = "tibco.alias.".length();
Properties p = new Properties();
p.load(new FileInputStream(file));
Properties modified = new Properties();
for (Iterator<Object> i = p.keySet().iterator(); i.hasNext();) {
String key = (String) i.next();
String value = p.getProperty(key);
if (key.length() != 0 && value.length() != 0) {
if (key.startsWith("tibco.alias."))
key = key.substring(start);
modified.setProperty(key, value);
value = value.replace("\\", "/");
File f = new File(value);
boolean fileOK = false;
try {
f = f.getCanonicalFile();
fileOK = f.exists();
}
catch (Exception exc) {}
if (!fileOK)
System.err.println("The file '"+f.getPath()+"' specified for alias '"+key+"' "+
" was not found.");
}
}
AEApplication.getAEApplication().setFileAliases(modified);
}
};
System.out.println("**** Processing aliases ****");
temp.updateAliases(new File(aliasFileLocation));
aliases = temp;
Properties props = createDocumentPropertiesForFile(projectLocation);
props.put("ae.forimport", "true");
props.put("ae.silent", "true");
System.out.println("**** Creating ActiveEnterprise document (serviceagents below may contain Java Instance instead of WebService with WSDL document) ****");
FileSystemVFileFactoryResourceStore storeFileFactory = new FileSystemVFileFactoryResourceStore(props) {
@Override
public void loadResourceFrom(DesignerFolder parent, VFile file)
throws Exception {
try {
String uri = file.getFullURI();
AEResource res = (AEResource) provider.getObject(uri, AEResource.class, true);
res.initializeNameFrom(file.getName());
parent.addResource(res);
}
catch (Exception e) { /* ignore unknown files */}
}
};
doc = new AEDocument(storeFileFactory, false);
doc.getResourceStore().insureFoldersAreInitialized();
ValidateProjectAction action = new ValidateProjectAction(doc);
System.out.println("**** Validating (you may see output from Java static initializers here) ****");
DesignerError errors[] = action.runValidation(false);

if (errors != null && errors.length > 0) {
printStatistics(errors);

ListHashMap errorTypeToErrorList = new ListHashMap();
int i = 0;
for (int max = errors.length; i < max; i++) {
String type = errors[i].getType();
if (type == null)
type = "ae.error.default.type";
errorTypeToErrorList.put(type, errors[i]);
}

for (@SuppressWarnings("rawtypes") Iterator iter = errorTypeToErrorList.keySet().iterator(); iter.hasNext(); ) {
String type = (String) iter.next();
@SuppressWarnings("unchecked")
List<DesignerError> errList = errorTypeToErrorList.getList(type);
StringBuffer sb = new StringBuffer();

if (!type.equals("ae.error.unused.gvardefn.type")) {
boolean ignoreXpathInGV = type.equals("ae.error.missing.gvardefn.type");
String categName = ResourceManager.manager.getString(type+".displayName");
sb.append("\n**** ["+type+"] "+categName+" ****\n\n");

int hits = 0;
for (int k=0; k < errList.size(); k++) {
DesignerError err = (DesignerError) errList.get(k);
String summ = err.getSummary();
String msg = err.getMessage();

if (ignoreXpathInGV && (msg.contains("[") || msg.contains("*")))
continue;
if (summ.contains("The resource") && summ.contains("is unknown"))
continue; /* Ignore not BW artifact files inside repository */
String uri = err.getAssociatedResource()!=null ? err.getAssociatedResource().getURI() : null;
if (uri!=null && uri.toLowerCase().startsWith("/test"))
continue; /* Ignore not valid unit tests */

sb.append("  "+summ+": "+msg+"\n");
if (uri != null)
sb.append("      <"+uri+">\r\n");
sb.append("\r\n");

if (type.contains("warning"))
warningCount++;
else
errorCount++;
hits++;
}
if (hits>0) {
System.err.println(sb);
}
}
}
if (errorCount==0)
System.out.println(ResourceManager.manager.getString("ae.validate.project.noerrors"));
}
cleanup();
}

@Override
public void cleanup() {
if (aliases != null)
aliases.resetAliases();
if (doc != null) {
try {
doc.close();
}
catch (Exception exp) {}
}
}

@Override
public void checkForErrors() {
if (DesignerError.getErrorCount() != 0) {
DesignerError errs[] = DesignerError.getErrorLog();
int i = 0;
for (int max = errs.length; i < max; i++) {
DesignerError err = errs[i];
System.err.println("[REPO ERROR] "+err.getSummary());
System.err.println((new StringBuilder()).append("\t")
.append(err.getMessage()).toString());
}
cleanup();
System.exit(1);
}
}

private void printStatistics(DesignerError errors[]) {
String s = ResourceManager.manager.getString(
"ae.validate.project.dialog.summary.status.text",
String.valueOf(errorCount), String.valueOf(warningCount));
System.out.println("**** Validation stats *****\n"+s);
}

public final static void main(String[] args) throws Exception {
String projectPath = null;
String aliasPath = null;
String[] cmdline = new String[args.length+3];
for (int k = 0; k < args.length; k++)
cmdline[k] = args[k];
for (int i=0; i < args.length; i++) {
if (args[i].equals("-a")) {
aliasPath = args[i+1];
if (projectPath==null)
projectPath = args[i+2];
}
else
projectPath = args[i];
}
if (projectPath==null || aliasPath==null ||
projectPath.trim().length()==0 || aliasPath.trim().length()==0) {
System.out.println("Usage: projectPath -a aliases.txt");
System.exit(0);
}
new ValidateBwProject().execute(projectPath, aliasPath);
System.exit(0);
}
}

How to add support for StartTls in Spring LDAP

import com.sun.jndi.ldap.LdapClient;

public class Binding extends LdapContextSource {

private SearchControls sc = new SearchControls();
private String base = null;
private static boolean requireServerCertInTrustedStore = false;
private static boolean traceAttributesModification = !true;
private boolean useStartTlsAuth = false;

private final static String[] CLEAR_PARAMS_FOR_STARTTLS = { ".security",
"java.naming.ldap.factory.socket", "com.sun.jndi.ldap.connect.pool"
};

private Hashtable<String,Object> prepareParamsForStartTls(@SuppressWarnings("rawtypes") Hashtable env) {
Hashtable<String,Object> props = new Hashtable<String, Object>();
for (Object k : env.keySet()) {
String key = k.toString();
int hitCount = 0;
for (int i = 0; i < CLEAR_PARAMS_FOR_STARTTLS.length; i++) {
if (key.contains(CLEAR_PARAMS_FOR_STARTTLS[i])) {
hitCount++;
break;
}
}
if (hitCount == 0)
props.put(key, env.get(key));
}
props.put("java.naming.ldap.version", "3");
props.put(Context.SECURITY_PROTOCOL, "plain");
return props;
}

@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
protected DirContext getDirContextInstance(Hashtable env)
throws NamingException {

Hashtable<String,Object> params = prepareParamsForStartTls(env); /* strip pooling and security */
params.put("java.naming.ldap.attributes.binary", "objectSid objectGUID");
final LdapContext ctx = (LdapContext) NamingManager.getInitialContext(params);

StartTlsResponse tlsResp = null;
if (useStartTlsAuth) {
tlsResp = (StartTlsResponse) ctx.extendedOperation(new StartTlsRequest());
tlsResp.setHostnameVerifier(TolerantSSLSocketFactory.TOLERANT_HOSTNAME_VERIFIER);
try {
tlsResp.negotiate(TolerantSSLSocketFactory.INSTANCE);
}
catch (Throwable t) {
throw new NamingException("SSL negotiation for LDAP StartTLS failed: "+t);
}
}
ctx.getEnvironment().putAll(env); /* do not use addToEnvironment because it would reconnect without StartTls */
try {
Field fClient = ctx.getClass().getDeclaredField("clnt");
fClient.setAccessible(true);
LdapClient cli = (LdapClient) fClient.get(ctx);
/* We are over secure channel, need to authenticate now */
Method mAuthenticate = cli.getClass().getDeclaredMethod("authenticate", new Class[] {
boolean.class, String.class, Object.class, int.class, String.class, Control[].class, Hashtable.class
});
mAuthenticate.setAccessible(true);
mAuthenticate.invoke(cli, false, env.get(Context.SECURITY_PRINCIPAL).toString(),
env.get(Context.SECURITY_CREDENTIALS), 3, "simple", null, env);
}
catch (Exception e) {
Throwable t = e;
while (t.getCause() != null)
t = t.getCause();
throw new NamingException("Cannot authenticate due to error: "+t.getMessage());
}
return new InitialLdapContext(env, null) {
@Override
protected Context getDefaultInitCtx() throws NamingException {
/* cache context */
return ctx;
}
};
}

public static boolean isRequiredServerCertInTrustedStore() {
return requireServerCertInTrustedStore;
}

public static void setRequireServerCertInTrustedStore(
boolean requireServerCertInTrustedStore) {
Binding.requireServerCertInTrustedStore = requireServerCertInTrustedStore;
}

public static void setTraceAttributesModification(
boolean traceAttributesModification) {
Binding.traceAttributesModification = traceAttributesModification;
}

public static boolean getTraceAttributesModification() {
return traceAttributesModification;
}

public Binding(String userLogin, String userPassword, String serverUrl, String searchBase) throws Exception {

useStartTlsAuth = serverUrl.startsWith("ldap(s)");
if (useStartTlsAuth) {
serverUrl = "ldap"+serverUrl.substring(7);
setPooled(false);
}
setUserDn(userLogin);
setPassword(userPassword);
setBase(searchBase);
setReferral("follow");
setUrl(serverUrl);
setPooled(true);

HashMap<String,String> props = new HashMap<String, String>();
props.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");
props.put(Context.SECURITY_AUTHENTICATION, "none");
props.put(Context.SECURITY_AUTHENTICATION, "simple");
props.put(Context.SECURITY_CREDENTIALS, userLogin);
props.put(Context.SECURITY_PRINCIPAL, userPassword);

if (!requireServerCertInTrustedStore) {
props.put("java.naming.ldap.factory.socket", TolerantSSLSocketFactory.class.getName());
}
setCacheEnvironmentProperties(true);
setBaseEnvironmentProperties(props);
setAuthenticationStrategy(new SimpleDirContextAuthenticationStrategy());
afterPropertiesSet();
sc.setSearchScope(SearchControls.SUBTREE_SCOPE);
sc.setCountLimit(0);
sc.setReturningAttributes(null);
sc.setReturningObjFlag(true);
}

public SearchControls getSearchControls() {
return sc;
}

public String getSearchBase() {
return this.base != null ? base : "";
}
}

Unescape Unicode (for Tibco dealing with international data)

ByteArrayOutputStream repacked = new ByteArrayOutputStream();
byte[] buff = (in+"\0\0\0\0\0\0").getBytes(); //padding to make McCabe happy
for (int i=0; i < buff.length-6; ) {
    if (buff[i] == '\\' && buff[i+1] == 'u') {
        String t = new String(new int[] { Integer.decode("0x"+new String(buff, i+2, 4)) }, 0, 1);
        for (int k=0; k < t.length(); k++) {
            int cp = t.codePointAt(k);
            if (!Character.isIdentifierIgnorable(cp))
                repacked.write(t.getBytes("utf-8"));
        }
        i+=6;
    }
    else {
        repacked.write(buff, i, 1);
        i++;
    }
}
in = repacked.toString(encoding);

How to copy EMS datastore for interDC transport? Ought to be _fast_.

#ifndef _GNU_SOURCE
#define _GNU_SOURCE 
#endif
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <malloc.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string.h>
#include <dlfcn.h>
#include <execinfo.h>
#include <pthread.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <sys/mman.h>
#include <sys/sendfile.h>
#include <errno.h>
#include <time.h>
#include <sys/syscall.h>
#include <inttypes.h>
#ifdef AIO_VIA_SYSCALL
#include <linux/aio_abi.h>
#else
#include <libaio.h>
#endif
#include <list>

#define SYSOUTF(msg, params...) do { char buff[1024]; memset(buff, 0, 1024); snprintf(buff, 1024, msg, params); write(1,buff,strlen(buff)); } while(0)
#define SYSOUT(msg) write(1, msg, strlen(msg))

#define ONE_MB (1024*1024)
#define ONE_GB (1024L*ONE_MB)

#define STRATEGY_MMAP_WRITE  0
#define STRATEGY_MMAP_BOTH   1
#define STRATEGY_SERIAL      2
#define STRATEGY_SENDFILE    3
#define STRATEGY_AIO         4

static const char* STRATEGY_DESC[] = { "MMAP+WRITE", "MMAPx2+MEMCPY", "READ+WRITE", "SENDFILE", "AIO" };

static struct timespec __ts, __ts_res;
static clockid_t __clock = CLOCK_REALTIME;

static int strategy = STRATEGY_MMAP_WRITE;
static int fd_in = -1, fd_out = -1, err_in = 0, err_out = 0;
static void *addr = NULL, *addr2 = NULL;
static long size_in = 0, blk_size = 0;
static int warn_mem_params_not_optimal = 1;

#define TIMER_START() do { timer_start(&__ts,__clock); } while(0)
#define TIMER_STOP() do { timer_stop(&__ts,__clock); clock_getres(__clock,&__ts_res); } while(0)
#define PRINT_WALL_TIME() do { long msec = __ts.tv_nsec/1000000L; SYSOUTF("[INFO] WALL TIME: %ldsec %ldmsec %ldusec +- %ldnsec\n",(long)__ts.tv_sec,msec,(__ts.tv_nsec - msec*1000000L)/1000L,__ts_res.tv_nsec); } while(0)

static inline int timer_start(struct timespec *ts, clockid_t clid) {
    int res = clock_gettime(clid,ts);
 if (res!=0) {
  int e = errno;
  SYSOUTF("Failed timer_start due to '%s'\r\n", strerror(e));
 }
    return res;
}

static inline int timer_stop(struct timespec *ts, clockid_t clid) {
    struct timespec ts1;
    int res = clock_gettime(clid,&ts1);
    if (res!=0) {
        int e = errno;
        SYSOUTF("Failed timer_stop due to '%s'\r\n", strerror(e));
  return res;
    }
    ts->tv_sec = ts1.tv_sec - ts->tv_sec;
    ts->tv_nsec = ts1.tv_nsec - ts->tv_nsec;
    if (ts->tv_nsec < 0) {
  ts->tv_nsec += 1000000000L;
        ts->tv_sec--;
    }
    return 0;
}

long get_rmem_limit() {
    struct rlimit rlimit;
    int res = getrlimit(RLIMIT_MEMLOCK, &rlimit);
    if (res == 0) {
  return (long)rlimit.rlim_max;
    }
    return -1L;
}

int open_source_file(char *path) {
    int res = -1;
    int extra_flags = strategy == STRATEGY_AIO ? O_DIRECT : 0;
    do {
  res = open(path, O_LARGEFILE | O_RDONLY | extra_flags);
    }
    while (res == -1 && errno == EINTR);    
    return res;
}

int open_dest_file(char *path) {
    int res = -1;
    int extra_flags = strategy == STRATEGY_AIO ? O_DIRECT : 0;
    do {
  res = open(path, O_CREAT | O_LARGEFILE | O_RDWR | extra_flags, 0666);
    }
    while (res == -1 && errno == EINTR);
    return res;
}

long get_file_size(int fd, long *block_size) {
    struct stat stat;
    TIMER_START();
    int res = fstat(fd, &stat);
    if (res == 0) {
  if (block_size != NULL)
   *block_size = stat.st_blksize;
  return stat.st_size;
    }
    return -1L;
}

int allocate_file(int fd, long size) {
    return fallocate(fd, 0, 0, size);
}

void preload_file_advise(int fd, long size, off_t base_offset) {
 int res = posix_fadvise(fd, base_offset, size, POSIX_FADV_WILLNEED | POSIX_FADV_SEQUENTIAL);
 if (res != 0) {
  int e = errno;
  SYSOUTF("[WARN] POSIX fadvise returned error '%s'\r\n", strerror(e));
 }
}

void* mmap_file(int fd, long size, int mode, off_t base_offset = 0) {
    preload_file_advise(fd, size, base_offset);
    int PERF_FLAGS[] = { MAP_HUGETLB | MAP_LOCKED, MAP_HUGETLB, MAP_LOCKED, 0 };
    void *addr = NULL;
    int i = 0;
    for (; i < 4; i++) {
  addr = mmap(NULL, size, mode == 0 ? PROT_READ : PROT_WRITE, MAP_SHARED | MAP_POPULATE | PERF_FLAGS[i], fd, base_offset);
  if (addr!= (void*)-1)
   break;
    }
    if (addr == (void*)-1)
  return NULL;
    if (i!=0 && warn_mem_params_not_optimal) {
  SYSOUT("[WARN] Memory access is not optimal (no huge TLB & pages locked into RAM)\r\n");
  warn_mem_params_not_optimal = 0;
    }
    int res = posix_madvise(addr, size, POSIX_MADV_WILLNEED | POSIX_MADV_SEQUENTIAL);
    if (res != 0) {
  int e = errno;
  SYSOUTF("[WARN] POSIX madvise returned error '%s'\r\n", strerror(e));
    }
    res = mlock(addr, size);
    if (res != 0) {
     int e = errno;
     SYSOUTF("[WARN] POSIX mlock returned error '%s'\r\n", strerror(e));
    }
    return addr;
}

int write_file(int fd, void *buff, long len, long base_off = 0) {
    long to_be_written = len;
    long off = 0;
    do {
  long res = pwrite(fd, buff, to_be_written, off + base_off);
  if (res > 0) {
   to_be_written -= res;
   off += res;
   if (to_be_written == 0L)
    break;
  }
  else if (res < 0 && errno != EINTR)
   break;
 }
    while (errno == EINTR);
    TIMER_STOP();
    if (errno == 0)
  fdatasync(fd); 
    return to_be_written == 0 ? 0 : errno;
}    

#define CLEANUP_EXIT(code) { if (addr) munmap(addr, size_in); if (addr2) munmap(addr2, size_in); if (fd_in!=-1) close(fd_in); if (fd_out!=-1) close(fd_out); exit(code); } while(0)
#define CLEANUP_MMAP(size) { if (addr) { munmap(addr, size); addr = NULL; } if (addr2) { munmap(addr2, size); addr2 = NULL; } } while(0)

void open_files(char **argv) {
    long mlock_max = get_rmem_limit();
    SYSOUTF("[INFO] RLIMIT_MEMLOCK is %ldKB,\r\n", mlock_max/1024);
    if (mlock_max < 128*1024*1024) {
  SYSOUTF("[WARN] Please adjust memlock in /etc/security/limits.conf to something big like %d (KB), which is 32GB.\r\n", 32*ONE_MB);
    }
    fd_in = open_source_file(argv[1]); err_in = errno;
    if (fd_in == -1) {
  SYSOUTF("[ERROR] Cannot open source file %s due to error '%s'\r\n", argv[1], strerror(err_in));
  CLEANUP_EXIT(-1);
    }
    fd_out = open_dest_file(argv[2]); err_out = errno;
    if (fd_out == -1) {
  SYSOUTF("[ERROR] Cannot open destination file %s due to error '%s'\r\n", argv[2], strerror(err_out));
  CLEANUP_EXIT(-2);
    }    
    size_in = get_file_size(fd_in, &blk_size); err_in = errno;
    if (size_in == -1L) {
  SYSOUTF("[ERROR] Unable determine file size for %s due to error '%s'\r\n", argv[1], strerror(err_in));
  CLEANUP_EXIT(-3);
    }
    SYSOUTF("[INFO] File size for %s is %ld, block size is %ld.\r\n", argv[1], size_in, blk_size);
    if (allocate_file(fd_out, size_in) != 0) {
  err_out = errno;
  SYSOUTF("[ERROR] Cannot allocate file %s due to error '%s'\r\n", argv[2], strerror(err_out));
  CLEANUP_EXIT(-4);
    }
}

#define MMAP_IN  0
#define MMAP_OUT 1

#ifndef MMAP_CHUNK_SIZE
#define MMAP_CHUNK_SIZE  (128*1024*1024)
#endif

void mmap_copy(char **argv) {
 for (long base = 0; base < size_in; base += MMAP_CHUNK_SIZE) {
  long size = size_in - base;
  if (size > MMAP_CHUNK_SIZE)
   size = MMAP_CHUNK_SIZE;
  else if (size <= 0)
   break;
   
  addr = mmap_file(fd_in, size, MMAP_IN, base);
  if (addr == NULL) {
   err_in = errno;
   SYSOUTF("[ERROR] Cannot load file %s into memory due to error '%s'\r\n", argv[1], strerror(err_in));
   CLEANUP_EXIT(-5);
  }
  if (strategy == STRATEGY_MMAP_WRITE) {
   int res = write_file(fd_out, addr, size, base);
   if (res!=0) {
    SYSOUTF("[ERROR] Could not write file %s due to error '%s'\r\n", argv[2], strerror(res));
    CLEANUP_EXIT(-6);
   }
  }
  else if (strategy == STRATEGY_MMAP_BOTH) {
   addr2 = mmap_file(fd_out, size, MMAP_OUT, base);
   if (addr2 == NULL) {
    err_out = errno;
    SYSOUTF("[ERROR] Cannot load file %s into memory due to error '%s'\r\n", argv[2], strerror(err_out));
    CLEANUP_EXIT(-7);
   }
   else {
    memcpy(addr2, addr, size);
    TIMER_STOP();
    if (msync(addr2, size, MS_SYNC) != 0) {
     err_out = errno;
     SYSOUTF("[ERROR] Cannot sync file %s due to error '%s'\r\n", argv[2], strerror(err_out));
     CLEANUP_EXIT(-8);
    }
   }
  }
  CLEANUP_MMAP(size);
 }
 TIMER_STOP();
}

int sendfile_copy() {
 err_out = 0;
 long to_write = size_in;
 long written_once = 0;
 while (to_write > 0) {
  written_once = sendfile(fd_out, fd_in, NULL, to_write);
  if (written_once == -1) {
   err_out = errno;
   if (err_out != EAGAIN) {
    SYSOUTF("[ERROR] Function sendfile is not supported '%s'\r\n", strerror(err_out));
    return -1;
   }
   else
    continue;
  }
  else {
   to_write -= written_once;
  }
 }
 if (to_write != 0) {
  SYSOUTF("[ERROR] Function sendfile written too small data: %ld vs %ld, error=%s\r\n", size_in - to_write, size_in, strerror(err_out));
  return -1;
 }
 TIMER_STOP();
 return 1;    
}    

#ifndef COPY_BLOCK_SIZE
#define COPY_BLOCK_SIZE 4096
#endif

void serial_copy() {
    char buff[COPY_BLOCK_SIZE];
    long in = 0;
    long out = 0;
    long cnt = 0;
    preload_file_advise(fd_in, size_in, 0);
    do {
  in=read(fd_in, (void*)buff, COPY_BLOCK_SIZE);
  if (in>0) {
   out = in;
   do {
    cnt = write(fd_out, buff, out);
    err_out = errno;
    if (cnt < 0 && err_out != EINTR) {
     SYSOUTF("[ERROR] Broken writing to output file '%s'\r\n", strerror(err_out));
     CLEANUP_EXIT(-10);
    }
    else if (cnt > 0) {
     out -= cnt;
    }
   }
   while (out > 0);
  }
  else {
   if (errno == 0) {
    if (in==0)
     continue;
    else
     break;
   }
   if (errno != EINTR) {
    err_in = errno;
    SYSOUTF("[ERROR] Broken reading of input file '%s'\r\n", strerror(err_in));
    CLEANUP_EXIT(-9); 
   }
   else
    continue;
  }
    }
    while (in > 0); 
 TIMER_STOP();
}

#define PG_SIZE 4096
#define AIO_MAX_NR 1024

#ifdef AIO_VIA_SYSCALL
inline int io_setup(unsigned nr, aio_context_t *ctxp) {
 return syscall(__NR_io_setup, nr, ctxp);
}
inline int io_destroy(aio_context_t ctx) {
 return syscall(__NR_io_destroy, ctx);
}
inline int io_submit(aio_context_t ctx, long nr, struct iocb **cbp) {
 return syscall(__NR_io_submit, nr, cbp);
}
inline int io_cancel(aio_context_t ctx, struct iocb *cb, struct io_event *result) {
 return syscall(__NR_io_cancel, cb, result);
}
inline int io_getevents(aio_context_t ctx, long min_nr, long nr, struct io_event *events, struct timespec *timeout) {
 return syscall(__NR_io_getevents, min_nr, nr, events, timeout);
}
inline void io_prep_pread(struct iocb *cb, int fd, void *buff, long size, long offset) {
 cb->aio_fildes = fd;
 cb->aio_lio_opcode = IOCB_CMD_PREAD;
 cb->aio_buf = (uint64_t)buff;
 cb->aio_offset = offset;
 cb->aio_nbytes = size;
}
#else
#define IOCB_CMD_PREAD  0
#define IOCB_CMD_PWRITE  1
#define IOCB_CMD_FSYNC  2
#define IOCB_CMD_FDSYNC  3
#define IOCB_CMD_NOOP  6
#define aio_context_t io_context_t
#endif

void aio_copy() {
 int chunks = (size_in + (PG_SIZE/2) - 1) / PG_SIZE; /* with padding */
 SYSOUTF("[INFO] AIO copy needs %d chunks\r\n", chunks);
 aio_context_t ctx = 0;
 struct iocb* cbs = (struct iocb*)malloc(AIO_MAX_NR * sizeof(struct iocb));
 struct iocb** cbps = (struct iocb**)malloc(AIO_MAX_NR * sizeof(struct iocb*));
 struct iocb** wcbps = (struct iocb**)malloc(AIO_MAX_NR * sizeof(struct iocb*));
 struct io_event* ioevents = (struct io_event*)malloc(AIO_MAX_NR * sizeof(struct io_event));

 if (cbs==NULL || cbps==NULL || wcbps==NULL || ioevents==NULL) {
  int ret = errno;
  SYSOUTF("[ERROR] Broken malloc '%s'\r\n", strerror(ret));
  CLEANUP_EXIT(-12);
 } 
 
 int ret = io_setup(AIO_MAX_NR, &ctx);
 if (ret < 0) {
  ret = errno;
  SYSOUTF("[ERROR] Broken io_setup for %d chunks '%s'\r\n", chunks, strerror(ret));
  CLEANUP_EXIT(-11); 
 }
 void *bigbuff = malloc(AIO_MAX_NR*PG_SIZE);
#define DELETE_MEM do { if (bigbuff!=NULL) free(bigbuff); free(cbs); free(cbps); free(wcbps); free(ioevents); io_destroy(ctx); } while(0)
 if (bigbuff == NULL) {
  ret = errno;
  SYSOUTF("[ERROR] Buffer allocation failed for input file: '%s'\r\n", strerror(ret));
  DELETE_MEM;
  CLEANUP_EXIT(-12); 
 }
 
 int rounds = (chunks + AIO_MAX_NR/2 - 1) / AIO_MAX_NR; /* with padding */
#define AIO_GROUP_TO_OFFSET(i) (long(i) * PG_SIZE * AIO_MAX_NR)
#define AIO_STEP_TO_OFFSET(i) (long(i) * PG_SIZE)
 for (int i=0; i < AIO_MAX_NR; i++) {
  memset(&cbs[i], 0, sizeof(struct iocb));
  cbps[i] = &cbs[i];
 }

 for (int group=0; group < rounds; group++) {
  int step = 0;
  for (; step < AIO_MAX_NR; step++) {
   long curr_off = AIO_GROUP_TO_OFFSET(group) + AIO_STEP_TO_OFFSET(step);
   long nbytes = size_in - curr_off;
   if (nbytes <= 0) /* we are inside padding */
    break;
   if (nbytes > PG_SIZE)
    nbytes = PG_SIZE;
 
   long pbuff = (long)bigbuff + AIO_STEP_TO_OFFSET(step);
   io_prep_pread(&cbs[step], fd_in, (void*)pbuff, nbytes, curr_off);
  }
  if (step == 0)
   break;
  int todo = io_submit(ctx, step, cbps);
  if (todo != step) {
   ret = errno;
   SYSOUTF("[ERROR] Broken io_submit, group=%d, queued=%d; '%s'\r\n", group, todo >= 0 ? todo : -1, todo < 0 ? strerror(ret) : "failure for given index");
   DELETE_MEM;
   CLEANUP_EXIT(-13); 
  }
  
  int wi = 0;
  while (todo > 0) {
   ret = io_getevents(ctx, 1, todo, ioevents, NULL);
   if (ret < 0) {
    ret = errno;
    SYSOUTF("[ERROR] Broken io_getevents '%s'\r\n", strerror(ret));
    DELETE_MEM;
    CLEANUP_EXIT(-14); 
   }
   else if (ret > 0) {
    todo -= ret;
    int to_write = 0;
    for (int i=0; i < ret; i++) {
     struct iocb *iocbp = (struct iocb*) ioevents[i].obj;
     if (iocbp->aio_lio_opcode == IOCB_CMD_PREAD) {
      iocbp->aio_fildes = fd_out;
      iocbp->aio_lio_opcode = IOCB_CMD_PWRITE;
      wcbps[wi++] = iocbp;
      to_write++;
     }
    }
    if (to_write > 0) {
     int e = io_submit(ctx, to_write, &wcbps[wi - to_write]); /* move along linearly */
     if (e != to_write) {
      ret = errno;
      SYSOUTF("[ERROR] Broken io_submit for write '%s'\r\n", e < 0 ? strerror(ret) : "failure for given index");
      DELETE_MEM;
      CLEANUP_EXIT(-15); 
     }  
     else
      todo += to_write;
    }
   }
  }
 }
 DELETE_MEM;
 TIMER_STOP();
}

int main(int argc, char **argv) {

 if (argc < 3) {
  SYSOUTF("Usage: %s SRC_FILE DST_FILE --strategy={mmap,serial,sendfile,aio,default}\r\n", argv[0]);
  return 0;
 }
 if (argc == 4 && strncmp(argv[3], "--strategy=mmap", 15) == 0)
  strategy = STRATEGY_MMAP_BOTH;
 else if (argc == 4 && strncmp(argv[3], "--strategy=serial", 17) == 0)
  strategy = STRATEGY_SERIAL;
 else if (argc == 4 && strncmp(argv[3], "--strategy=sendfile", 19) == 0)
  strategy = STRATEGY_SENDFILE;
 else if (argc == 4 && strncmp(argv[3], "--strategy=aio", 14) == 0)
  strategy = STRATEGY_AIO;
    
 open_files(argv);
 if (strategy == STRATEGY_SERIAL)
  serial_copy();
 else if (strategy == STRATEGY_SENDFILE)
  sendfile_copy();
 else if (strategy == STRATEGY_AIO)
  aio_copy();
 else
  mmap_copy(argv);
    
 SYSOUTF("\r\n[INFO] Success. Strategy was %s\r\n", STRATEGY_DESC[strategy]);
 PRINT_WALL_TIME();
 CLEANUP_EXIT(0);
 return 0;
}

Jak oglądać telewizję gdy niemowlak śpi?

Transmiter Bluetooth Audio (RCA/Jack) podłączony do tunera albo telewizora + słuchawki BT. Urządzenie można zasilać z portu USB telewizora/tunera lub ładowarki do komórki. Gdy transmiter jest podłączony do tunera należy go włączyć i wyciszyć telewizor za pomocą pilota. W przypadku podłączenia do telewizora należy wpiąć w TV kabel Jack, telewizor wyłączy wtedy głośniki. Na zwykłych słuchawkach bez kodeka aptx (wirtualnie wszystkie dostępne na Allegro słuchawki bezprzewodowe robione przez Chińczyków wystawiających towar na alibaba.com) jest lag do 1s, ale można z tym żyć. Porządne słuchawki zapewniające dobrą synchronizację są co najmniej 2 razy droższe. Rozwiązanie jest bardzo fajne, ale ma minus - jest jednoosobowe. Transmiter FM i dwie sztuki słuchawek z radiem mają sporo gorszą jakość dźwięku.

EMS DS tracer

#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string.h>
#include <dlfcn.h>
#include <execinfo.h>
#include <pthread.h>
#include <list>

const char interp[] __attribute__((section(".interp"))) = "/lib64/ld-linux-x86-64.so.2";

typedef int (*open_fn)(const char *path, int flags, ...);
typedef ssize_t (*write_fn)(int fd, const void *buf, size_t count);
typedef ssize_t (*writev_fn)(int fd, const struct iovec *iov, int count);
typedef ssize_t (*pwritev_fn)(int fd, const struct iovec *iov, int count, off_t offset);
typedef int (*ftruncate_fn)(int fd, off_t length);

static open_fn _open = NULL;
static open_fn _open64 = NULL;
static write_fn _write = NULL;
static writev_fn _writev = NULL;
static pwritev_fn _pwritev = NULL;
static ftruncate_fn _ftruncate = NULL;
static void *handle_libc = NULL;
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

#define SYSOUTF(msg, params...) do { char buff[1024]; memset(buff, 0, 1024); snprintf(buff, 1024, msg, params); _write(1,buff,strlen(buff)); } while(0)
#define SYSOUT(msg) _write(1, msg, strlen(msg))

void __attribute__((constructor)) __init(void)
{
    if (handle_libc==NULL)
handle_libc = dlopen ("/lib64/libc.so.6", RTLD_NOW);
    if (handle_libc!=NULL) {
_open = (open_fn) dlsym(handle_libc, "open");
_open64 = (open_fn) dlsym(handle_libc, "open64");
_write = (write_fn) dlsym(handle_libc,"write");
_writev = (writev_fn) dlsym(handle_libc,"writev");
_pwritev = (pwritev_fn) dlsym(handle_libc,"pwritev");
_ftruncate = (ftruncate_fn) dlsym(handle_libc, "ftruncate");
SYSOUT("libemscompanion init completed\r\n");
    }
    else {
exit(-1);
    }
}

void __attribute__((destructor)) __fini(void)
{
    if (handle_libc!=NULL)
dlclose(handle_libc);
}

extern "C" {

    int open(const char *path, int flags, ...) {
va_list vargs;
va_start(vargs, flags);
mode_t mode = va_arg(vargs, mode_t);
va_end(vargs);
int ret = _open(path, flags, mode);
SYSOUTF("Open %s = %d\r\n", path, ret);
return ret;
    }
   
    int open64(const char *path, int flags, ...) {
va_list vargs;
va_start(vargs, flags);
mode_t mode = va_arg(vargs, mode_t);
va_end(vargs);
int ret = _open(path, flags, mode);
SYSOUTF("Open %s = %d\r\n", path, ret);
return ret;
    }
   
    ssize_t write(int fd, const void *buf, size_t count) {
off_t curr = lseek(fd, 0, SEEK_CUR);
SYSOUTF("Writing fd = %d, addr = %d, buff = %x, size = %d\r\n", fd, curr, buff, count);
ssize_t ret = _write(fd, buf, count);
return ret;
    }
}

extern "C" void _main(int argc, char **argv) {
    __init();
    SYSOUT("libEMSCompanion 0.0.1\r\n");
    exit(0);
}

EMS File Datastore as a sequential tx log

Writing fd = 9, addr = 24834560, buff = 8c405540, size = 512 (chunk header, const offset, every 1910272 bytes)
Writing fd = 9, addr = 22405120, buff = 8c405710, size = 861696 (full message with headers and body)
Writing fd = 9, addr = 23266816, buff = 8c405710, size = 861696
Writing fd = 9, addr = 26744832, buff = 8c405540, size = 512
Writing fd = 9, addr = 24128512, buff = 8c405710, size = 861696
Writing fd = 9, addr = 24990208, buff = 89c01820, size = 512
Writing fd = 9, addr = 1024, buff = 89c01820, size = 512
Writing fd = 9, addr = 1536, buff = 89c01820, size = 512
Writing fd = 9, addr = 24990720, buff = 89c01820, size = 512 (Client ACK record)
Writing fd = 9, addr = 24991232, buff = 89c01820, size = 512
Writing fd = 9, addr = 24991744, buff = 89c01820, size = 512
Writing fd = 9, addr = 24992256, buff = 89c01820, size = 512
Writing fd = 9, addr = 24992768, buff = 89c01820, size = 512
Writing fd = 9, addr = 24993280, buff = 89c01820, size = 512
Writing fd = 9, addr = 24993792, buff = 89c01820, size = 512
Writing fd = 9, addr = 24994304, buff = 89c01820, size = 512
Writing fd = 9, addr = 24994816, buff = 89c01820, size = 512
Writing fd = 9, addr = 24995328, buff = 89c01820, size = 512
Writing fd = 9, addr = 24995840, buff = 89c01820, size = 512
Writing fd = 9, addr = 24996352, buff = 89c01820, size = 512
Writing fd = 9, addr = 24996864, buff = 89c01820, size = 512
Writing fd = 9, addr = 24997376, buff = 89c01820, size = 512
Writing fd = 9, addr = 24997888, buff = 89c01820, size = 512
Writing fd = 9, addr = 24998400, buff = 89c01820, size = 512
Writing fd = 9, addr = 24998912, buff = 89c01820, size = 512
Writing fd = 9, addr = 24999424, buff = 89c01820, size = 512
Writing fd = 9, addr = 24999936, buff = 89c01820, size = 512
Writing fd = 9, addr = 25000448, buff = 89c01820, size = 512
Writing fd = 9, addr = 25000960, buff = 89c01820, size = 512
Writing fd = 9, addr = 25001472, buff = 89c01820, size = 512
Writing fd = 9, addr = 25001984, buff = 89c01820, size = 512
Writing fd = 9, addr = 25002496, buff = 89c01820, size = 512
Writing fd = 9, addr = 25003008, buff = 89c01820, size = 512
Writing fd = 9, addr = 25003520, buff = 89c01820, size = 512
Writing fd = 9, addr = 25004032, buff = 79ffa800, size = 512
Writing fd = 9, addr = 25004544, buff = 79ffa750, size = 512
Writing fd = 9, addr = 512, buff = 8c405710, size = 861696 (full message with headers and body, space reused)

Full messages and ACK records are laid out sequentially and interwoven. ACK doesn't clean up message header. This design under heavy load may lead to excessive datastore usage and need for frequent compaction maintenance.

JMS headers with almost nothing set take 213 bytes. Minus 28 bytes for JMSMessageID it is 185 bytes on disk.