<commit_before>using System;
using System.Collections.Generic;
using System.Threading;
using System.Net.Sockets;
using System.Text; 

namespace Gearman
{

	public class Worker
	{
		
		public delegate byte[] taskdelegate(byte[] indata); 
		
		private TcpClient conn; 
		private string hostname; 
		private int port;
		private Dictionary<string, taskdelegate> methodMap;
		
		
		public Worker (string host)
		{
			this.port = 4730;
			this.hostname = host;
			this.init();
		}
		
		public Worker (string host, int port)
		{
			this.hostname = host; 
			this.port = port; 
			this.init();
		}
		
		public void registerFunction(string taskname, taskdelegate function)
		{
			methodMap[taskname] = function;
			RequestPacket r = new RequestPacket();
			r.Type = PacketType.CAN_DO;
			r.setData(taskname);
			NetworkStream stream = conn.GetStream();
			stream.Write(r.ToByteArray(),0, r.length());
			stream.Flush();
		}
		                        
		public void init()
		{
			try { 
				methodMap = new Dictionary<string, taskdelegate>();
				conn = new TcpClient(hostname, port);
			} catch (Exception e) { 
				Console.WriteLine("Error initializing Worker: " + e.ToString());
			}
		}
		
		public void workLoop()
		{
			Thread t = new Thread(new ThreadStart(run));
			t.Start();
		}
		
		public void run()
		{
 
			while(true)
			{
				nextPacket();
				Console.WriteLine("Sleeping for 2 seconds");
				Thread.Sleep(2000);
			}
				
		}

		
		private void nextPacket()
		{
			byte[] buf = new byte[1];
			byte[] pdata = null, message = null;
			int totalBytes = 0; 
			int bytesRead;
			bool pktDone = false; 
			int poffset = 0; 
			int messagesize = -1; 
			
			ASCIIEncoding encoder = new ASCIIEncoding();
			NetworkStream stream = conn.GetStream(); 
			
			// Grab job from server (if available)
			Console.WriteLine("Checking for job...");
			RequestPacket rp = new RequestPacket();
			rp.Type = PacketType.GRAB_JOB;
			conn.GetStream().Write(rp.ToByteArray(), 0, rp.length());
			conn.GetStream().Flush();
			pktDone = false; 
			message = new byte[65536];
			messagesize = -1; 
			totalBytes = 0; 
			
 			while (!pktDone) 
			{
				try {
					bytesRead = stream.Read(buf, 0, 1);
					message[totalBytes++] = buf[0];
				
					if(totalBytes == 12) { 
						// Check byte count
						byte[] sizebytes = message.Slice(8,12); 
						
						if(BitConverter.IsLittleEndian)
							Array.Reverse(sizebytes);
						
						messagesize = BitConverter.ToInt32(sizebytes, 0);
						Console.WriteLine("Packet is another {0} bytes", messagesize);
					}
					
					if(messagesize != -1 && messagesize == (totalBytes - 12))
					{
						Console.WriteLine("Done parsing message");
						pktDone = true;
						pdata = new byte[totalBytes+1];
						Array.Copy(message, pdata, totalBytes);
					} 

				} catch (Exception e) { 
					Console.WriteLine("Exception reading data: {0}", e.ToString());
				}
			} 
			
			Console.WriteLine("Finished processing packet!");
			if(pdata != null) 
			{
				Packet p = new Packet(pdata);
				p.Dump();
			
			
				if (p.Type == PacketType.JOB_ASSIGN)
				{
					Console.WriteLine("Assigned job: {0}", p.JobHandle);
					byte[] payload = p.RawData;
					bool gotTask = false;
					string task = "";
					int taskoffset = 0; 
					
					for(int i = 0; i < payload.Length; i++)
					{
						if (payload[i] == 0)
						{
							if(!gotTask)
							{
								Console.WriteLine("Got zero byte and no task yet...");
								task = encoder.GetString(payload, 0, i);
								taskoffset = i;
								gotTask = true; 		
							} else {
								Console.WriteLine("Task: {0}", task); 
								byte[] result    = methodMap[task](payload.Slice(taskoffset, i));
								byte[] jobhandle = ASCIIEncoding.UTF8.GetBytes(p.JobHandle);
								
								byte[] d = new byte[jobhandle.Length + result.Length + 1];
								
								Array.Copy(jobhandle, d, jobhandle.Length);
								Array.Copy(result, 0, d, jobhandle.Length + 1, result.Length);
							
								RequestPacket workresult = new RequestPacket();
								workresult.Type = PacketType.WORK_COMPLETE; 
								workresult.setJobData(d);
								
								workresult.Dump();
								
								conn.GetStream().Write(workresult.ToByteArray(), 0, workresult.length());
								conn.GetStream().Flush();
							}
							
						}
						
						
					}
					
				}
				
				if (p.Type == PacketType.NO_JOB)
				{
					Console.WriteLine("Nothing to do!");
				} 
			}
		}
		
	}
}
<commit_msg>[project @ 88b8980c5d5dd2de6eddceb31eb0956f2af7d89a] Implemented Connection class, added a stopWorkLoop() method, and reaplced a lot of the connection / packet parsing code with cleaner Connection-based methods.<commit_after>using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;
using System.Net.Sockets;
using System.Text; 
using log4net;

namespace Gearman
{

	public class Worker
	{

		public delegate byte[] taskdelegate(Packet jobPacket, Connection c); 

		private Connection c; 
		private Dictionary<string, taskdelegate> methodMap;
		private Thread t; 
		public static readonly ILog Log = LogManager.GetLogger(typeof(Worker));

		
		public Worker()
		{
			methodMap = new Dictionary<string, taskdelegate>();
		}
		
		public Worker (string host, int port) : this()
		{
			this.c = new Connection(host, port);
		}
		
		public Worker (string host) : this(host, 4730)
		{		}

		public void registerFunction(string taskname, taskdelegate function)
		{
			methodMap[taskname] = function;
			RequestPacket r = new RequestPacket();
			r.Type = PacketType.CAN_DO;
			r.setData(taskname);
			c.sendPacket(r);
		}
		                        
		public void workLoop()
		{
			t = new Thread(new ThreadStart(run));
			t.Start();
		}
		
		public void stopWorkLoop()
		{
			if (t != null)
			{
				t.Abort();
			}
		}
		
		
		public void run()
		{
			while(true)
			{
				checkForJob();
				Log.DebugFormat("Sleeping for 2 seconds");
				Thread.Sleep(2000);
			}		
		}

		
		private void checkForJob()
		{
			ASCIIEncoding encoder = new ASCIIEncoding(); 
			// Grab job from server (if available)
			Log.DebugFormat("Checking for job...");
			RequestPacket rp = new RequestPacket();
			rp.Type = PacketType.GRAB_JOB;
			
			c.sendPacket(rp);
			
			Packet response = c.getNextPacket();
			
			if (response.Type == PacketType.JOB_ASSIGN)
			{
				Log.DebugFormat("Assigned job: {0}", response.JobHandle);
				byte[] payload = response.Data;
				bool gotTask = false;
				string task = "";
				
				for(int i = 0; i < payload.Length; i++)
				{
					if (payload[i] == 0 || i == payload.Length - 1)
					{
						if(!gotTask)
						{
							Log.DebugFormat("Got zero byte and no task yet...");
							task = encoder.GetString(payload, 0, i);
							gotTask = true; 		
						} else {
							Log.DebugFormat("Task: {0}", task); 
							byte[] result      = methodMap[task](response, c);
							string resultText  = encoder.GetString(result);
							
							Log.DebugFormat("Result: " + resultText);
							
							RequestPacket workresult = new RequestPacket(PacketType.WORK_COMPLETE, result, response.JobHandle);
							c.sendPacket(workresult);
							
							workresult.Dump();
							
						}	
					}	
				}
			} else if (response.Type == PacketType.NO_JOB) {
				Log.DebugFormat("Nothing to do!");
			} 
		}
		
	}
}

{'fraction_non_alphanumeric': 0.12602852441031268, 'fraction_numerical': 0.011793746571585299, 'mean_word_length': 2.4449692961738307, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, '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': '3339064', 'n_tokens_mistral': 3233, 'n_tokens_neox': 2286, 'n_words': 545}