<commit_before>from mpi4py import MPI
import subprocess
import sys
import os
import signal
import socket
import threading
import time
from Queue import Queue, Empty
import struct
import posix_ipc
import sysv_ipc
import mmap
import shmemUtils

DEBUG = True

#MATLAB_BIN = '/usr/local/packages/MATLAB/R2011b/bin/matlab'
MATLAB_BIN = os.environ['MATLAB'] + '/bin/matlab'
PRESTO_DIR = os.environ['PRESTO']
CHECKIN_TAG = 1
DO_WORK_TAG = 2
KILL_TAG    = 3
RESULTS_TAG = 4

WORKER_PORT = 11110
MASTER_PORT = 11112

JQKEY = 37408
RQKEY = 48963

TMPFS_PATH = '/dev/shm/'

# Set to True if using "NETWORK" mode
messageQueue = False

def launch_matlab(queue, worker_dict):
  # Without GUI:
  presto_mpath = os.path.join(PRESTO_DIR, 'presto_setpath.m')

  p = subprocess.Popen(MATLAB_BIN + ' -nodesktop -nosplash -r \"run ' + presto_mpath + ';' + 
                       worker_dict + '\"', shell=True)
  # With GUI:
  #p = subprocess.Popen(MATLAB_BIN + ' -desktop -r \"'+ worker_dict + 
  #                     '\"', shell=True)
  queue.put('running')
  p.communicate()
  # send kill signal to all workers
  q.put('kill')

def launch_java(queue, worker_dict, java_app):
  p = subprocess.Popen('java ' + java_app, shell=True)
  queue.put('running')
  p.communicate()
  # send kill signal to all workers
  q.put('kill')


def master_shmem_initiailization():
  #try:
  #  semaphore = posix_ipc.Semaphore('MATSEM')
  #  semaphore.unlink()
  #except:
  #  pass 
  #try: 
  #  memory = posix_ipc.SharedMemory('MAT2SHM')
  #  memory.unlink()
  #except:
  #  pass
  try: 
    jq = sysv_ipc.MessageQueue(JQKEY)
    jq.remove()
  except:
    pass
  try: 
    # Results Queue (rq)
    rq = sysv_ipc.MessageQueue(RQKEY)
    rq.remove()
  except:
    pass
 
def worker_shmem_initiailization():
  try:
    memory = posix_ipc.SharedMemory('SHM2MAT')
    memory.unlink()
  except:
    pass
  try:
    semaphore = posix_ipc.Semaphore('PYSEM')
    semaphore.unlink()
  except:
    pass

# Get the user id for the file objects
def get_uid():
  uid = subprocess.check_output('id')
  uid = uid.split()[0]
  lft = uid.index('=') + 1
  rht = uid.index('(')
  uid = uid[lft:rht]
  return uid


# Initialize mpi4py:
mpiComm = MPI.COMM_WORLD
size = mpiComm.Get_size()
rank = mpiComm.Get_rank()
name = MPI.Get_processor_name()
srank = str(rank)
interface = sys.argv[1]

print "Process %d of %d launched on %s." % (rank, size-1, name)


# MASTER PROCESS:
if (rank==0):
  master_status = MPI.Status()
  data = {}
  for i in range(1,size):
    data[i] = mpiComm.recv(source=i, tag=CHECKIN_TAG)

  worker_dict = 'matlabyrinth_workers = struct(\'hosts\', {'
  for i in range(1,size):
    worker_dict = worker_dict + '\'' + data[i]['name'] + '\', '
  worker_dict = worker_dict + '}, \'ranks\', {'
  for i in range(1,size):
    worker_dict = worker_dict + str(data[i]['rank']) + ', '
  worker_dict = worker_dict + '})'

  q = Queue()
  if interface == 'matlab':
    print 'sysv is:' + str(len(sys.argv))
    if len(sys.argv) == 3:
      worker_dict = worker_dict + ';' + sys.argv[2]
      print 'worker_dict:' + worker_dict
    t = threading.Thread(target=launch_matlab, args=(q, worker_dict))
  elif interface == 'java':
    java_app = sys.argv[2]
    t = threading.Thread(target=launch_java, args=(q, worker_dict, java_app))
  t.start()

  master_shmem_initiailization()<commit_msg>changing name to presto_workers<commit_after>from mpi4py import MPI
import subprocess
import sys
import os
import signal
import socket
import threading
import time
from Queue import Queue, Empty
import struct
import posix_ipc
import sysv_ipc
import mmap
import shmemUtils

DEBUG = True

#MATLAB_BIN = '/usr/local/packages/MATLAB/R2011b/bin/matlab'
MATLAB_BIN = os.environ['MATLAB'] + '/bin/matlab'
PRESTO_DIR = os.environ['PRESTO']
CHECKIN_TAG = 1
DO_WORK_TAG = 2
KILL_TAG    = 3
RESULTS_TAG = 4

WORKER_PORT = 11110
MASTER_PORT = 11112

JQKEY = 37408
RQKEY = 48963

TMPFS_PATH = '/dev/shm/'

# Set to True if using "NETWORK" mode
messageQueue = False

def launch_matlab(queue, worker_dict):
  # Without GUI:
  presto_mpath = os.path.join(PRESTO_DIR, 'presto_setpath.m')

  p = subprocess.Popen(MATLAB_BIN + ' -nodesktop -nosplash -r \"run ' + presto_mpath + ';' + 
                       worker_dict + '\"', shell=True)
  # With GUI:
  #p = subprocess.Popen(MATLAB_BIN + ' -desktop -r \"'+ worker_dict + 
  #                     '\"', shell=True)
  queue.put('running')
  p.communicate()
  # send kill signal to all workers
  q.put('kill')

def launch_java(queue, worker_dict, java_app):
  p = subprocess.Popen('java ' + java_app, shell=True)
  queue.put('running')
  p.communicate()
  # send kill signal to all workers
  q.put('kill')


def master_shmem_initiailization():
  #try:
  #  semaphore = posix_ipc.Semaphore('MATSEM')
  #  semaphore.unlink()
  #except:
  #  pass 
  #try: 
  #  memory = posix_ipc.SharedMemory('MAT2SHM')
  #  memory.unlink()
  #except:
  #  pass
  try: 
    jq = sysv_ipc.MessageQueue(JQKEY)
    jq.remove()
  except:
    pass
  try: 
    # Results Queue (rq)
    rq = sysv_ipc.MessageQueue(RQKEY)
    rq.remove()
  except:
    pass
 
def worker_shmem_initiailization():
  try:
    memory = posix_ipc.SharedMemory('SHM2MAT')
    memory.unlink()
  except:
    pass
  try:
    semaphore = posix_ipc.Semaphore('PYSEM')
    semaphore.unlink()
  except:
    pass

# Get the user id for the file objects
def get_uid():
  uid = subprocess.check_output('id')
  uid = uid.split()[0]
  lft = uid.index('=') + 1
  rht = uid.index('(')
  uid = uid[lft:rht]
  return uid


# Initialize mpi4py:
mpiComm = MPI.COMM_WORLD
size = mpiComm.Get_size()
rank = mpiComm.Get_rank()
name = MPI.Get_processor_name()
srank = str(rank)
interface = sys.argv[1]

print "Process %d of %d launched on %s." % (rank, size-1, name)


# MASTER PROCESS:
if (rank==0):
  master_status = MPI.Status()
  data = {}
  for i in range(1,size):
    data[i] = mpiComm.recv(source=i, tag=CHECKIN_TAG)

  worker_dict = 'presto_workers = struct(\'hosts\', {'
  for i in range(1,size):
    worker_dict = worker_dict + '\'' + data[i]['name'] + '\', '
  worker_dict = worker_dict + '}, \'ranks\', {'
  for i in range(1,size):
    worker_dict = worker_dict + str(data[i]['rank']) + ', '
  worker_dict = worker_dict + '})'

  q = Queue()
  if interface == 'matlab':
    print 'sysv is:' + str(len(sys.argv))
    if len(sys.argv) == 3:
      worker_dict = worker_dict + ';' + sys.argv[2]
      print 'worker_dict:' + worker_dict
    t = threading.Thread(target=launch_matlab, args=(q, worker_dict))
  elif interface == 'java':
    java_app = sys.argv[2]
    t = threading.Thread(target=launch_java, args=(q, worker_dict, java_app))
  t.start()

  master_shmem_initiailization()
{'fraction_non_alphanumeric': 0.14200298953662183, 'fraction_numerical': 0.012855007473841554, 'mean_word_length': 3.4966397849462365, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6160373', 'n_tokens_mistral': 2665, 'n_tokens_neox': 2556, 'n_words': 564}