Author: k.indika.maligaspe@gmail.com

I recently had to a do an application that required me to sort a data set fast and efficiently. I opted to do a merge sort. Merge sort is a fast sorting algorithm which is always true with an average or worst case performance of O(n log n). What I really love about a merge sort is that it is a stable sorting algorithm.

After doing the sorting I realized that most people are unaware of how actually merge sort worked. Hence I decided to do this post so anyone can actually understand how merge sort worked behind the scene and give example codes using java and C of a merge sort on an integer array.

Merge Sort is a divide an concur sorting algorithm. The Array is divided repeatedly until we have only 2 arrays of only one element each. Then the elements in the two arrays are sorted and we work our way back up the hierarchy merging the arrays till we get a sorted array. You can get a very good description on the wiki site about merge sort.

If we take an array of 6 integers, say 5,12,3,3,9,18, we follow the process below

1. Break the array in to two arrays -> 5,12,3 and 3,9,18
2. Take each array and break it again -> 5 and 12,3
3. Break the array till you have only 1 element each -> 12 and 3
4. Now start sorting and merging – >3 and 12
   1. one layer up 3,12 and 1 -> 1,3,12
   2. one layer up 1,3,12 and 3,9,18 – > 1,3,3,9,12,18
5. Now you have a sorted array

Let me explain this now with a Java code.
Note that I did not use any specific java syntax such as array.length to keep the code as generic as possible. The code methods are self explanatory.

class merge_sort
{

    public int[] split_merge(int [] mainArr , int start, int end)
    {
	if ((end - start) < 2) {
	    return mainArr;
	}

	int middle = (start + end) / 2;

	int[] A = new int[middle - start];
	int[] B = new int[end - middle];

	A = copy_array(mainArr, start, middle,A);
	B = copy_array(mainArr, middle, end ,B);

        //Recursively Call split_merge till we have an 
        //array with 1 element
	A = split_merge(A,0,middle);
	B = split_merge(B,0 ,(end-middle));
        
        //merge our way back one layer up always.
	mainArr = merge(mainArr, A, B, start, middle, end);
	return mainArr;
    }

    public int[] merge(int[] array, int[] A, int[] B, int start, int middle, int end)
    {
	int i = 0 , j = 0, k = 0;
        //Loop till end of main array and copy the sorted elements.
	for (k=0;k<(end-start);k++){
	    if ( (i < middle)  && (A[i] < B[j]) ){
		    array[k] = A[i];
		    i=i+1;
	    }else{
		    array[k] = B[j];
		    j=j+1;
	   }
	}

	return array;
	
    }

    public int[] copy_array(int[] A, int start, int end, int[] B)
    {
	int i =start, j =0;
        while (i < end) {
	    B[j] = A[i];
	    j+=1;
	    i+=1;
	}
	return B;
    }
    public  static void  main(String args[])
    {
	int mainArr[] = {2,3,5,1,6,4,10};
	int length = 7;

	for(int i =0 ; i <mainarr.length; i++){
="" system.out.format("unsorted="" array[%d]="%d" \n="" ",="" i="" ,="" mainarr[i]);="" }="" mainarr="new" merge_sort().split_merge(mainarr,0,length);="" for(int="" ;="" <mainarr.length;="" system.out.format("sorted="" <="" pre="">

Now let’s look at the same as implemented using C language. Again I have tried to make it language agnostic as possible.

#include <stdio.h>
#include <string.h>

int *merge_sort(int A[],int length);
int *copy_array(int *A,int start , int end , int *B);
int *split_merge(int *A, int start, int end);
int *merge(int *arr, int *A, int *B, int start,int middle, int end);


int* copy_array(int *A, int start, int end,int *B)
{
 int i , j;
  j = 0;
  for ( i = start; i< end; i++){
    B[j] = A[i];
    j = j +1;
  }
  return B;
}

int *merge(int *arr , int *A, int *B, int start, int middle, int end)
{
  int i = 0;
  int j = 0;
  int k = 0;

  for(k = 0; k < end ; k++){
    if((i < middle) && (A[i] < B[j])){
      arr[k] = A[i];
      i = i +1 ;
    }else {
      arr[k] = B[j];
      j = j +1;
    }  
  }
  return arr;
}

int *split_merge(int *arr, int start, int end)
{
  if((end-start) < 2)
    return arr;
  int k;
  
  int middle = (end -start) /2;
  int size_a = middle - start;
  int size_b = end - middle;
  
  int A[size_a];
  int B[size_b];
  int *_aptr = copy_array(arr , 0 , middle , A);
  int *_bptr = copy_array(arr, middle , end, B);

  _aptr = split_merge(_aptr,0,middle);
  _bptr = split_merge(_bptr,0, (end - middle));

  arr = merge(arr , _aptr , _bptr , start , middle, end);
  return arr;
  
}

int *merge_sort(int A[], int length)
{
  int start  = 0;
  int end = length; //I put this to match the Java Code
  A = split_merge(A,start,end);
  return A;
}


int main(int argc, char argv[])
{
  int A[] = {100,50,200,1000,18,5300};
  int length = 6;
  int *B;

  for ( i = 0;i < length ; i++){
    printf("Unsorted Array[%d] : %d \n",i , A[i]);
  }

  int *aptr = merge_sort(A , length );
  int i;
 
  for ( i = 0;i < length ; i++){
    printf("Sorted Array[%d] : %d \n",i , aptr[i]);
  }
  
  return 0;
}

So I hope that this will help you next time when you have to sort a data set and you will choose merge as an option to your application.

import logging
import pika
import json
import time
from logging.handlers import RotatingFileHandler
from threading import Thread



class Publisher(Thread):

    def __init__(self, RABBITMQ_SETTINGS,LOG4PY_SETTINGS):
        Thread.__init__(self)       
        self.logger = logging.getLogger("Publisher.py")
        self.connection = None
        self.channel = None
        self._deliveries = []
        self._acked = 0
        self._nacked = 0
        self._message_number = 0
        self._stopping = False
        self.queue = 'alienvault_replicate'
        self.routing_key = 'alienvault_replicate'
        self.exchange = 'alienvault_replicate'
        self.message = None
        self.ready = False
        self._closing = False
        log4py_file  = LOG4PY_SETTINGS['log4py_file']
        log4py_log_level = LOG4PY_SETTINGS['log4py_log_level']
        self.PUBLISH_INTERVAL=0.1
        self.RABBITMQ_SETTINGS =RABBITMQ_SETTINGS

        if log4py_log_level == 'DEBUG':
           self.log_level = logging.DEBUG
        elif log4py_log_level == 'INFO': 
           self.log_level = logging.INFO
        elif log4py_log_level == 'WARN':
           self.log_level = logging.WARN
        elif log4py_log_level == 'ERROR': 
           self.log_level = logging.ERROR

        self.logger.setLevel(self.log_level)
        # create console handler and set level to debug
        rfh = RotatingFileHandler(filename=log4py_file, mode='a', maxBytes=100*1024*1024,backupCount=2)
        rfh.setLevel(self.log_level)
        # create formatter
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    
        # add formatter to ch
        rfh.setFormatter(formatter)
    
        # add ch to logger
        self.logger.addHandler(rfh)
        
        amqp_url = 'amqp://'+self.RABBITMQ_SETTINGS['user']+':'+self.RABBITMQ_SETTINGS['passwd']+'@'+self.RABBITMQ_SETTINGS['host']+':'+str('5672')+'/%2F'
        self._url = amqp_url
        
    def is_ready(self):
        return self.ready
        
    def set_message(self,message):
        self.message = message
        self.logger.info('Message set to publish to {0}'.format(self.message))
    
    
    def connect(self):
        self.logger.info('Connecting to %s', self._url)
        return pika.SelectConnection(pika.URLParameters(self._url),
                                     self.onconnection_open)
        
    def close_connection(self):
        self.logger.info('Closing connection')
        self._closing = True
        self.connection.close()

    def add_onconnection_close_callback(self):
        self.logger.info('Adding connection close callback')
        self.connection.add_on_close_callback(self.onconnection_closed)

    def onconnection_closed(self, connection, reply_code, reply_text):

        self.channel = None
        if self._closing:
            self.connection.ioloop.stop()
        else:
            self.logger.warning('Connection closed, reopening in 5 seconds: (%s) %s',
                           reply_code, reply_text)
            self.ready = False                   
            self.reconnect()


    def onconnection_open(self, unusedconnection):
        self.logger.info('Connection opened')
        self.add_onconnection_close_callback()
        self.openchannel()

    def reconnect(self):
        self.connection.ioloop.stop()
        self.connection = self.connect()
        self.connection.ioloop.start()

    def add_onchannel_close_callback(self):
        self.logger.info('Adding channel close callback')
        self.channel.add_on_close_callback(self.onchannel_closed)

    def onchannel_closed(self, channel, reply_code, reply_text):
        self.logger.warning('Channel was closed: (%s) %s', reply_code, reply_text)
        if not self._closing:
            self.connection.close()

    def onchannel_open(self, channel):
        self.logger.info('Channel opened')
        self.channel = channel
        self.add_onchannel_close_callback()
        self.setup_exchange(self.exchange)

    def setup_exchange(self, exchange_name):
        self.logger.info('Declaring exchange %s', exchange_name)
        self.channel.exchange_declare(self.on_exchange_declareok,
                                       exchange_name)

    def on_exchange_declareok(self, unused_frame):
        self.logger.info('Exchange declared')
        self.setup_queue(self.queue)

    def setup_queue(self, queue_name):
        self.logger.info('Declaring queue %s', queue_name)
        self.channel.queue_declare(self.on_queue_declareok, queue_name,durable=True)

    def on_queue_declareok(self, method_frame):
        self.logger.info('Binding %s to %s with %s',
                    self.exchange, self.queue, self.routing_key)
        self.channel.queue_bind(self.on_bindok, self.queue,
                                 self.exchange, self.routing_key)


    def publish_message(self):
        if self._stopping:
            return
        if self.message == None:
            return

        try:
            properties = pika.BasicProperties(delivery_mode = 1)
    
            self.channel.basic_publish(self.exchange, self.routing_key,
                                        self.message,
                                        properties)
            self.logger.info('Published message # %i', self._message_number)
        except Exception as err:
            import trace
            self.logger.info("Error in sending message ... {0}".format(err.message))
            self.ready = False

    def start_publishing(self):
        self.logger.info('Issuing consumer related RPC commands')
        self.ready = True
        self.publish_message()

    def on_bindok(self, unused_frame):
        self.logger.info('Queue bound')
        self.start_publishing()

    def closechannel(self):
        self.logger.info('Closing the channel')
        if self.channel:
            self.channel.close()

    def openchannel(self):
        self.logger.info('Creating a new channel')
        self.connection.channel(on_open_callback=self.onchannel_open)

    def run(self):
        self.connection = self.connect()
        self.connection.ioloop.start()


    def stop(self):
        self.logger.info('Stopping')
        self._stopping = True
        self.closechannel()
        self.close_connection()
        self.connection.ioloop.start()
        self.logger.info('Stopped')

if __name__ == '__main__':
    try:
        RABBITMQ_SETTINGS = {"user":"user","passwd":"pw","host":"xxx.xxx.xxx.xxx"}
        LOG4PY_SETTINGS = {"log4py_file":"./log4py.log","log4py_log_level":"INFO"}
        
        publisher = Publisher(RABBITMQ_SETTINGS,LOG4PY_SETTINGS)
        publisher.start()
        for i in range(1 , 100000):
            message = '{"message":"Hello Fellasss...."'+str(i)+'}'
            publisher.set_message(message)
            while not publisher.is_ready():
                time.sleep(.001)
            publisher.publish_message()
        publisher.stop()    
    except KeyboardInterrupt:
        publisher.stop()