Changeset 8707

Timestamp:

09/28/12 10:49:02 (12 years ago)

Author:

jkarder

Message:

#1712: added user queue used to schedule tasks

Location:

branches/HiveTaskScheduler

Files:

• HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.dbml (modified) (1 diff)
• HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.dbml.layout (modified) (1 diff)
• HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.designer.cs (modified) (4 diffs)
• HeuristicLab.Services.Hive.DataAccess/3.3/SQL Scripts/Initialize Hive Database.sql (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Convert.cs (modified) (2 diffs)
• HeuristicLab.Services.Hive/3.3/DataTransfer/UserPriority.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/HiveDao.cs (modified) (2 diffs)
• HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs (modified) (3 diffs)
• HeuristicLab.Services.Hive/3.3/Properties/Settings.Designer.cs (modified) (2 diffs)
• HeuristicLab.Services.Hive/3.3/Properties/Settings.settings (modified) (1 diff)
• HeuristicLab.Services.Hive/3.3/Scheduler/RoundRobinTaskScheduler.cs (modified) (3 diffs)
• HeuristicLab.Services.Hive/3.3/app.config (modified) (1 diff)

branches/HiveTaskScheduler/HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.dbml

@@ -206 +206 @@
     <Type Name="UserPriority">
       <Column Name="UserId" Type="System.Guid" DbType="UniqueIdentifier NOT NULL" IsPrimaryKey="true" CanBeNull="false" />
+      <Column Name="DateEnqueued" Type="System.DateTime" DbType="DateTime" CanBeNull="false" />
     </Type>
   </Table>

branches/HiveTaskScheduler/HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.dbml.layout

@@ -241 +241 @@
       </nodes>
     </associationConnector>
-    <classShape Id="f9e8867f-fd15-4a72-8ca4-4f02cd3f141f" absoluteBounds="4.125, 5.5, 2, 1.0016910807291666">
+    <classShape Id="f9e8867f-fd15-4a72-8ca4-4f02cd3f141f" absoluteBounds="4.125, 5.5, 2, 1.1939925130208327">
       <DataClassMoniker Name="/HiveDataContext/UserPriority" />
       <nestedChildShapes>
-        <elementListCompartment Id="ee41f516-7d9c-4a1d-a1b8-bbe00a6ffea8" absoluteBounds="4.14, 5.96, 1.9700000000000002, 0.44169108072916663" name="DataPropertiesCompartment" titleTextColor="Black" itemTextColor="Black" />
+        <elementListCompartment Id="ee41f516-7d9c-4a1d-a1b8-bbe00a6ffea8" absoluteBounds="4.14, 5.96, 1.9700000000000002, 0.63399251302083326" name="DataPropertiesCompartment" titleTextColor="Black" itemTextColor="Black" />
       </nestedChildShapes>
     </classShape>

branches/HiveTaskScheduler/HeuristicLab.Services.Hive.DataAccess/3.3/HiveDataContext.designer.cs

@@ -3 +3 @@
 // <auto-generated>
 //     This code was generated by a tool.
-//     Runtime Version:4.0.30319.269
+//     Runtime Version:4.0.30319.17929
 //
 //     Changes to this file may cause incorrect behavior and will be lost if
@@ -4433 +4433 @@
     private System.Guid _UserId;
     
+    private System.DateTime _DateEnqueued;
+    
     #region Extensibility Method Definitions
     partial void OnLoaded();
@@ -4439 +4441 @@
     partial void OnUserIdChanging(System.Guid value);
     partial void OnUserIdChanged();
+    partial void OnDateEnqueuedChanging(System.DateTime value);
+    partial void OnDateEnqueuedChanged();
     #endregion
     
@@ -4466 +4470 @@
     }
     
+    [global::System.Data.Linq.Mapping.ColumnAttribute(Storage="_DateEnqueued", DbType="DateTime")]
+    public System.DateTime DateEnqueued
+    {
+      get
+      {
+        return this._DateEnqueued;
+      }
+      set
+      {
+        if ((this._DateEnqueued != value))
+        {
+          this.OnDateEnqueuedChanging(value);
+          this.SendPropertyChanging();
+          this._DateEnqueued = value;
+          this.SendPropertyChanged("DateEnqueued");
+          this.OnDateEnqueuedChanged();
+        }
+      }
+    }
+    
     public event PropertyChangingEventHandler PropertyChanging;
     

branches/HiveTaskScheduler/HeuristicLab.Services.Hive.DataAccess/3.3/SQL Scripts/Initialize Hive Database.sql

@@ -123 +123 @@
 CREATE TABLE [UserPriority](
   [UserId] UniqueIdentifier NOT NULL,
+  [DateEnqueued] DateTime NOT NULL,
   CONSTRAINT [PK_UserPriority] PRIMARY KEY ([UserId])
   )

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Convert.cs

@@ -531 +531 @@
     public static DT.UserPriority ToDto(DB.UserPriority source) {
       if (source == null) return null;
-      return new DT.UserPriority() { UserId = source.UserId };
+      return new DT.UserPriority() { Id = source.UserId, DateEnqueued = source.DateEnqueued };
     }
     public static DB.UserPriority ToEntity(DT.UserPriority source) {
@@ -540 +540 @@
     public static void ToEntity(DT.UserPriority source, DB.UserPriority target) {
       if ((source != null) && (target != null)) {
-        target.UserId = source.UserId;
+        target.UserId = source.Id;
+        target.DateEnqueued = source.DateEnqueued;
       }
     }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/DataTransfer/UserPriority.cs

@@ -28 +28 @@
   public class UserPriority : HiveItem {
     [DataMember]
-    public Guid UserId { get; set; }
+    public DateTime DateEnqueued { get; set; }
   }
 }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/HiveDao.cs

@@ -259 +259 @@
         var entity = DT.Convert.ToEntity(dto);
         db.Jobs.InsertOnSubmit(entity);
+        if (!db.Jobs.Any(x => x.OwnerUserId == dto.OwnerUserId))
+          EnqueueUserPriority(new DT.UserPriority { Id = dto.OwnerUserId, DateEnqueued = dto.DateCreated });
         db.SubmitChanges();
         return entity.JobId;
@@ -863 +865 @@
       }
     }
+
+    public void EnqueueUserPriority(DT.UserPriority dto) {
+      using (var db = CreateContext()) {
+        var entity = db.UserPriorities.FirstOrDefault(x => x.UserId == dto.Id);
+        if (entity == null) db.UserPriorities.InsertOnSubmit(DT.Convert.ToEntity(dto));
+        else DT.Convert.ToEntity(dto, entity);
+        db.SubmitChanges();
+      }
+    }
     #endregion
 

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Interfaces/IHiveDao.cs

@@ -154 +154 @@
     #region UserPriority Methods
     IEnumerable<DT.UserPriority> GetUserPriorities(Expression<Func<UserPriority, bool>> predicate);
+    void EnqueueUserPriority(DT.UserPriority userPriority);
     #endregion
   }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Manager/HeartbeatManager.cs

@@ -23 +23 @@
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using HeuristicLab.Services.Hive.DataTransfer;
 using DA = HeuristicLab.Services.Hive.DataAccess;
@@ -28 +29 @@
 namespace HeuristicLab.Services.Hive {
   public class HeartbeatManager {
+    private const string MutexName = "HiveTaskSchedulingMutex";
+
     private IHiveDao dao {
       get { return ServiceLocator.Instance.HiveDao; }
@@ -63 +66 @@
         // assign new task
         if (heartbeat.AssignJob && slave.IsAllowedToCalculate && heartbeat.FreeCores > 0) {
-          var availableJobs = taskScheduler.Schedule(dao.GetWaitingTasks(slave));
-          if (availableJobs.Any()) {
-            var job = availableJobs.First();
-            if (AssignJob(slave, job))
-              actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, job.Id));
+          bool mutexAquired = false;
+          var mutex = new Mutex(false, MutexName);
+          try {
+            mutexAquired = mutex.WaitOne(Properties.Settings.Default.SchedulingPatience);
+            if (!mutexAquired)
+              DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling could not be aquired.");
+            else {
+              var availableJobs = taskScheduler.Schedule(dao.GetWaitingTasks(slave));
+              if (availableJobs.Any()) {
+                var job = availableJobs.First();
+                if (AssignJob(slave, job))
+                  actions.Add(new MessageContainer(MessageContainer.MessageType.CalculateTask, job.Id));
+              }
+            }
+          }
+          catch (AbandonedMutexException) {
+            DA.LogFactory.GetLogger(this.GetType().Namespace).Log("HeartbeatManager: The mutex used for scheduling has been abandoned.");
+          }
+          finally {
+            if (mutexAquired) mutex.ReleaseMutex();
           }
         }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Properties/Settings.Designer.cs

@@ -2 +2 @@
 // <auto-generated>
 //     This code was generated by a tool.
-//     Runtime Version:4.0.30319.269
+//     Runtime Version:4.0.30319.17929
 //
 //     Changes to this file may cause incorrect behavior and will be lost if
@@ -77 +77 @@
             }
         }
+        
+        [global::System.Configuration.ApplicationScopedSettingAttribute()]
+        [global::System.Diagnostics.DebuggerNonUserCodeAttribute()]
+        [global::System.Configuration.DefaultSettingValueAttribute("00:00:20")]
+        public global::System.TimeSpan SchedulingPatience {
+            get {
+                return ((global::System.TimeSpan)(this["SchedulingPatience"]));
+            }
+        }
     }
 }

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Properties/Settings.settings

@@ -21 +21 @@
       <Value Profile="(Default)">3.00:00:00</Value>
     </Setting>
+    <Setting Name="SchedulingPatience" Type="System.TimeSpan" Scope="Application">
+      <Value Profile="(Default)">00:00:20</Value>
+    </Setting>
   </Settings>
 </SettingsFile>

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/Scheduler/RoundRobinTaskScheduler.cs

@@ -20 +20 @@
 #endregion
 
+using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -31 +32 @@
 
     public IEnumerable<Task> Schedule(IEnumerable<Task> tasks, int count = 1) {
-      var userPriorities = dao.GetUserPriorities(x => true).ToArray();
+      if (!tasks.Any()) return Enumerable.Empty<Task>();
+
+      var userPriorities = dao.GetUserPriorities(x => true).OrderBy(x => x.DateEnqueued).ToArray();
 
       var jobs = new List<Job>();
       foreach (var userPriority in userPriorities) {
         var closureUserPriority = userPriority;
-        jobs.AddRange(dao.GetJobs(x => x.OwnerUserId == closureUserPriority.UserId));
+        jobs.AddRange(dao.GetJobs(x => x.OwnerUserId == closureUserPriority.Id));
       }
 
@@ -43 +46 @@
         job => job.Id,
         (task, job) => new { Task = task, Job = job })
-        .ToDictionary(k => k, v => false);
+        .OrderByDescending(x => x.Task.Priority)
+        .ToList();
 
       var scheduledTasks = new List<Task>();
       int priorityIndex = 0;
+
       if (count == 0 || count > taskJobRelations.Count) count = taskJobRelations.Count;
+
       for (int i = 0; i < count; i++) {
-        var defaultEntry = taskJobRelations.First(x => !x.Value); // search first task which is not included yet
-        var priorityEntries = taskJobRelations.Where(x => !x.Value && x.Key.Job.OwnerUserId == userPriorities[priorityIndex].UserId).ToArray(); // search for tasks with desired user priority
-        while (!priorityEntries.Any() && priorityIndex < priorityEntries.Length)
-          priorityEntries = taskJobRelations.Where(x => !x.Value && x.Key.Job.OwnerUserId == userPriorities[++priorityIndex].UserId).ToArray();
+        var defaultEntry = taskJobRelations.First(); // search first task which is not included yet
+        var priorityEntries = taskJobRelations.Where(x => x.Job.OwnerUserId == userPriorities[priorityIndex].Id).ToArray(); // search for tasks with desired user priority
+        while (!priorityEntries.Any() && ++priorityIndex < userPriorities.Length)
+          priorityEntries = taskJobRelations.Where(x => x.Job.OwnerUserId == userPriorities[priorityIndex].Id).ToArray();
         if (priorityEntries.Any()) { // tasks with desired user priority found
-          var priorityEntry = priorityEntries.OrderByDescending(x => x.Key.Task.Priority).ThenBy(x => x.Key.Task.DateCreated).First();
-          if (defaultEntry.Key.Task.Priority <= priorityEntry.Key.Task.Priority) {
-            taskJobRelations[priorityEntry.Key] = true;
-            scheduledTasks.Add(priorityEntry.Key.Task);
+          var priorityEntry = priorityEntries.OrderByDescending(x => x.Task.Priority).ThenBy(x => x.Job.DateCreated).First();
+          if (defaultEntry.Task.Priority <= priorityEntry.Task.Priority) {
+            taskJobRelations.Remove(priorityEntry);
+            scheduledTasks.Add(priorityEntry.Task);
             priorityIndex++;
-          } else { // there are other tasks with higher priority
-            taskJobRelations[defaultEntry.Key] = true;
-            scheduledTasks.Add(defaultEntry.Key.Task);
+          } else { // there are other tasks with higher priorities
+            taskJobRelations.Remove(defaultEntry);
+            scheduledTasks.Add(defaultEntry.Task);
           }
         } else {
-          taskJobRelations[defaultEntry.Key] = true;
-          scheduledTasks.Add(defaultEntry.Key.Task);
+          taskJobRelations.Remove(defaultEntry);
+          scheduledTasks.Add(defaultEntry.Task);
         }
       }
 
-      // dequeue and enqueue priorities [0, priorityIndex), i.e. {4,3,6,2,5,1} and priorityIndex = 3 -> {2,5,1,4,3,6}
-      // i don't know how to solve this yet ... (race conditions, locking?)
+      // requeue user priorities
+      if (priorityIndex < userPriorities.Length)
+        for (int i = 0; i < priorityIndex; i++) {
+          userPriorities[i].DateEnqueued = DateTime.Now;
+          dao.EnqueueUserPriority(userPriorities[i]);
+        }
 
       return scheduledTasks;

branches/HiveTaskScheduler/HeuristicLab.Services.Hive/3.3/app.config

@@ -26 +26 @@
                 <value>3.00:00:00</value>
             </setting>
+            <setting name="SchedulingPatience" serializeAs="String">
+                <value>00:00:20</value>
+            </setting>
         </HeuristicLab.Services.Hive.Properties.Settings>
     </applicationSettings>