using System;
using System.Collections;
using System.Data;
using System.Data.SqlClient;
using System.Threading;
using edu.jhu.Cas.Jobs;
using edu.jhu.Cas.Util;

namespace edu.jhu.Cas.Service {
    public class RunJobsThread : CJobsDBThread {
        static CasJobs cj = new CasJobs();
        static CasJobsAdmin cja = new CasJobsAdmin();
        const string EXECUTING = "executing";
        const string EXEC_CAP = "exec_cap";
        const string CONTEXT = "context";
     
        public override string ServiceName {
            get {
                return "Run Jobs Service";
            }
        }
        public override string ServiceDescription {
            get {
                return "Executes non-quick query jobs.";
            }
        }
        public override int DefaultPanicMinutes {get {return 5;}}

        private const string RUN_JOBS_QUERY = @"
/*
 * this statement returns two tables.
 * the first table is an ordered list of waiting jobs
 * the second table is a list of the number of jobs that should be started per queue
 */
" + QuickQueueNamesQuery + @"
/*
find all waiting jobs where the owners of those jobs
do not already have running queries in the same queue
the tuple as string thing is very nasty, but don't know an easier way to write it
ss2k does not have intersect or except
*/
select j.*
from (
	--all waiting jobs, and their users
	select cast(webservicesid as varchar(12))+'_'+target as idt,jobid  from jobs j1
	where j1.status = 0 
) a, jobs j
where a.idt not in (
	--users with jobs currently running in non-quick queues and the names of those queues
	select cast(webservicesid as varchar(12))+'_'+ target as idt from jobs j
	where j.status in (1,2) and 
    j.taskname not like '%"+SYSTEM.TASKNAME+@"%' and
	j.hostip not in (select * from @quickQueueNames)
) and a.jobid = j.jobid
and (outputtype = 'QUERY' or outputtype = 'HEARTBEAT')
order by j.vtimesubmit asc
/*
 * how many jobs to start per queue?
 */
select l.target as " + CONTEXT + @", l.execlimit as " + EXEC_CAP + @", r.executing as " + EXECUTING + @"
from (
	select distinct s.target, s.execlimit
	from servers s
	where s.name not in (select * from @quickQueueNames)
) l left outer join (
	select distinct s.target, count(*) as executing
	from servers s, jobs j
	where s.name not in (select * from @quickQueueNames)
	and ( j.status = 1 or j.status = 2) --running or cancelling
	and s.name = j.hostip
	group by s.target, s.execlimit
) r on l.target = r.target
";
        const string initDBQuery = @"
update jobs 
set status = 4, 
message = 'Sorry, your job did not complete successfully.  Please try again.'
where status = 1

update jobs 
set status = 2,
message = 'Job cancelled.'
where status = 2
";
        static Hashtable RunningJobs = new Hashtable();
        const int CYCLE_DELAY_SECONDS = 10;
        SqlCommand rjqCom = new SqlCommand();

        public RunJobsThread()
            : base() {
            base.con.Open();
            rjqCom.Connection = base.con;
            initDB();//do the initial clean-up
            rjqCom.CommandText = RUN_JOBS_QUERY;
            rjqCom.Prepare();//then compile this query
        }
        void initDB() {
            rjqCom.CommandText = initDBQuery;
            rjqCom.ExecuteNonQuery();
        }
        public static void KillRunningJob(long jobid) {
            try {
                if (RunningJobs.ContainsKey(jobid)) {
                    RunningJobThread rjt = (RunningJobThread)RunningJobs[jobid];
                    RunningJobs.Remove(jobid);
                    rjt.KillJob();
                } else {//this job is not running, must not have started or already been cancelled.  
                    Job[] job = cj.FindJob(ID.SERVICE_ID, "jobid = " + jobid);
                    if (job == null || job.Length == 0) return;
                    cja.CompleteCancel(job[0].WebServicesID, jobid, "Job cancelled.", job[0].Created_Table);
                }
            } catch (Exception e) {
                CJobsService.Log(e.ToString(), System.Diagnostics.EventLogEntryType.Error);
            }
        }
        Hashtable userAdded = new Hashtable();
        protected override void Run() {
            for (; ; ) {
                try {
                    Thread.Sleep(CYCLE_DELAY_SECONDS * 1000);
                    CycleComplete();
                    if (State == CJobsDBState.RUNNING) {
                        DataSet ds = base.BatchAdminQuery(rjqCom);
                        Console.WriteLine("PASS");
                        Console.ReadLine();
                        DataTable execCounts = ds.Tables[1]; //targets, used slots, available slots
                        Job[] waitingJobs = DataConverter.dataTable2Job(ds.Tables[0]);//waiting jobs where users don't already have a job in that queue
                        if (ds.Tables[0].Rows.Count == 0) continue;
                        int started, executing, exec_cap;
                        string current_context;
                        foreach (DataRow dra in execCounts.Rows) {
                            current_context = dra[CONTEXT].ToString().ToUpper();
                            started = 0;
                            exec_cap = (int)dra[EXEC_CAP];
                            executing = dra[EXECUTING] is DBNull ? 0 : (int)dra[EXECUTING];
                            int start_this_many = exec_cap - executing;
                            if (start_this_many < 1) continue;//this context is full, next please
                            userAdded.Clear();
                            for (int i = 0; i < waitingJobs.Length; i++) {//add jobs until slots are full
                                if (started >= start_this_many) break;
                                if (waitingJobs[i].Target.ToUpper() == current_context
                                    && !userAdded.ContainsKey(waitingJobs[i].WebServicesID)) {
                                    (new RunningJobThread(waitingJobs[i])).StartJobThread();
                                    userAdded.Add(waitingJobs[i].WebServicesID, null);//only add single user per pass
                                    ++started;
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    CJobsService.Log(e.ToString(), System.Diagnostics.EventLogEntryType.Error);
                }
            }
        }
        protected override void Halt() { }
        class RunningJobThread {
            static CasJobs cj = new CasJobs();
            public Job job;
            public DRStatement qObject = null;
            public Thread JobThread;
            public RunningJobThread(Job j) {
                this.job = j;
            }
            public void StartJobThread() {
                JobThread = new Thread(new ThreadStart(this.RunQuery));
                if (RunningJobs.ContainsKey(job.JobID)) return;//job is already running, ignore
                RunningJobs.Add(job.JobID, this);
                JobThread.Start();
            }
            public void KillJob() {
                if (qObject != null)
                    qObject.Cancel(null, null);
                cja.CompleteCancel(job.WebServicesID, job.JobID, "Job cancelled.", job.Created_Table);
            }
            void RunQuery() {
                try {
                    Console.WriteLine("Starting " + job.JobID);
                    Server[] servers = null;
                    try {
                        servers = cj.getVirtualHostAndMyDBServer(job.WebServicesID, job.Target, job.Estimate);
                        qObject = new DRStatement(job, servers[1], servers[0], false, job.HostIP);
                    } catch (Exception e) {
                        try {
                            ToughJobs.FailJob(0, job.WebServicesID, job.JobID, e.Message, job.Created_Table);
                        } catch (Exception ee) {
                            CJobsService.Log(ee.ToString(), System.Diagnostics.EventLogEntryType.Error);
                        }
                    }
                    try {
                        if (servers != null && qObject != null)
                            cj.RunJob(job, servers[1], qObject);
                    } catch { }//exceptions in RunJob are handled by RunJob
                } catch (Exception e) {
                    CJobsService.Log(e.ToString(), System.Diagnostics.EventLogEntryType.Error);
                } finally {
                    if (job != null && RunningJobs.ContainsKey(job.JobID))
                        RunningJobs.Remove(job.JobID);
                }
            }
        }
    }
}