PeriodicTask is dynamic and save behavior changed
        See: https://github.com/zakird/celerybeat-mongo/commit/dfbbd20edde91134b57f5406d0ce4eac59d6899b
329726
def save(self, commit=True, **kwargs):
        '''
        
        '''
        if not self.instance:
            self.instance = self.model_class()  # Force populate_obj in super()
        return super(PeriodicTaskForm, self).save(commit, **kwargs)
917212
def sync_config_tasks(self):
        

        tasks_by_hash = {_hash_task(t): t for t in self.config_tasks}

        for task in self.all_tasks:
            if tasks_by_hash.get(task["hash"]):
                del tasks_by_hash[task["hash"]]
            else:
                self.collection.remove({"_id": task["_id"]})
                log.debug("Scheduler: deleted %s" % task["hash"])

        # What remains are the new ones to be inserted
        for h, task in tasks_by_hash.items():
            task["hash"] = h
            task["datelastqueued"] = datetime.datetime.fromtimestamp(0)
            if task.get("dailytime"):
                # Because MongoDB can store datetimes but not times,
                # we add today's date to the dailytime.
                # The date part will be discarded in check()
                task["dailytime"] = datetime.datetime.combine(
                    datetime.datetime.utcnow(), task["dailytime"])
                task["interval"] = 3600 * 24

                # Avoid to queue task in check() if today dailytime is already passed
                if datetime.datetime.utcnow().time() > task["dailytime"].time():
                    task["datelastqueued"] = datetime.datetime.utcnow()

            self.collection.find_one_and_update({"hash": task["hash"]}, {"$set": task}, upsert=True)
            log.debug("Scheduler: added %s" % task["hash"])
1204647
function defaultWorkflow(opts) {
  var tasks;
  var paths;
  var runnableTasks;
  var runnablePaths;

  return execP('git status')
    .then(function() {
      return getUnchanged(opts.globs);
    })
    .spread(function(a, b) {
      tasks = a;
      paths = b;
      console.log(
        'Executing: ' + '`git stash && git add . && git stash`.\n'.italic
      );
      return gitCmds.safeStashAll();
    })
    .then(function() {
      console.log(
        'Repo reverted to clean state. Looking at previous screenshots...\n'
      );
      console.log(
        'If anything goes wrong'.bold + ', simply run ' +
        '`git stash pop && git reset HEAD . && git stash pop --index` '.italic +
        'to restore your changes.'
      );
      // TODO: not good enough. Warn earlier about no task. Before stash
      var res = filterRunnables(tasks, paths, opts.taskName);
      runnableTasks = res[0];
      runnablePaths = res[1];

      return runTasks(writeScreenshots, opts, runnableTasks, runnablePaths);
    })
    .catch(function(e) {
      if (e.name === 'OperationalError' &&
          e.message.indexOf('Not a git repository') > -1) {
        return Promise.reject(new Error(notGitMsg));
      }

      return gitCmds
        .safeUnstashAll()
        .then(function() {
          return Promise.reject(e);
        });
    })
    .then(function() {
      console.log(
        'Executing: ' +
        '`git stash pop && git reset HEAD . && git stash pop --index`.\n'.italic
      );
      return gitCmds.safeUnstashAll();
    })
    .then(function() {
      console.log(
        'Repo back to modified state. Getting new screenshots and comparing ' +
        'them with previously recorded ones...\n'
      );
      return runTasks(compareScreenshots, opts, runnableTasks, runnablePaths);
    })
    .then(function() {
      return Promise.map(paths, function(dir, i) {
        return Promise.map(tasks[i], function(task) {
          var p = path.join(
            dir,
            consts.HUXLEY_FOLDER_NAME,
            task.name + consts.TASK_FOLDER_SUFFIX
          );
          return removeDirIfOk(p);
        });
      });
    });
}
49899
def save_object!(object, objectID = nil, request_options = {})
      res = save_object(object, objectID, request_options)
      wait_task(res['taskID'], WAIT_TASK_DEFAULT_TIME_BEFORE_RETRY, request_options)
      res
    end
1199072
function patchMacroTask(obj, funcName, metaCreator) {
    var setNative = null;
    function scheduleTask(task) {
        var data = task.data;
        data.args[data.cbIdx] = function () {
            task.invoke.apply(this, arguments);
        };
        setNative.apply(data.target, data.args);
        return task;
    }
    setNative = patchMethod(obj, funcName, function (delegate) { return function (self, args) {
        var meta = metaCreator(self, args);
        if (meta.cbIdx >= 0 && typeof args[meta.cbIdx] === 'function') {
            return scheduleMacroTaskWithCurrentZone(meta.name, args[meta.cbIdx], meta, scheduleTask);
        }
        else {
            // cause an error by calling it directly.
            return delegate.apply(self, args);
        }
    }; });
}
277800
function loadSync(updateSeconds) {
  if(!this.config.savefile) return null;
  try {
    var lastSync = fs.readFileSync(this.config.savefile, { encoding: 'utf8' });
    var json = JSON.parse(lastSync);
    if(typeof json !== 'object' || !json) return null;
    // We now update the remaining seconds for each task, if set.
    if(updateSeconds === true) {
      var now = new Date().getTime();
      for(var taskName in json) {
        var lastSave = json[taskName];
        json[taskName] = Math.floor((now - lastSave) / 1000);
      }
    }
    return json;
  } catch(e) {
    // no previous dumps;
    return null;
  }
}
974375
def save(self, force=False, uuid=False, **kwargs):
        

        if not self._initialized_with_doc and not force:
            raise Exception("Cannot save a document not initialized from a Python dict. This might remove fields from the DB!")

        self._initialized_with_doc = False

        if '_id' not in self:
            if uuid:
                self['_id'] = str("%s-%s" % (self.mongokat_collection.__class__.__name__, uuid4()))

        return self.mongokat_collection.save(self, **kwargs)
416159
public Status do_GET_active_tasks(Database _db, String _docID, String _attachmentName) {
        // http://wiki.apache.org/couchdb/HttpGetActiveTasks

        String feed = getQuery("feed");
        longpoll = "longpoll".equals(feed);
        boolean continuous = !longpoll && "continuous".equals(feed);

        String session_id = getQuery("session_id");

        ChangeListener listener = new ChangeListener() {

            ChangeListener self = this;

            @Override
            public void changed(ChangeEvent event) {

                Map<String, Object> activity = getActivity(event.getSource());

                // NOTE: Followings are not supported by iOS. We might remove them in the future.
                if (event.getTransition() != null) {
                    // this adds data to the response based on the trigger.
                    activity.put("transition_source", event.getTransition().getSource());
                    activity.put("transition_destination", event.getTransition().getDestination());
                    activity.put("trigger", event.getTransition().getTrigger());
                    Log.d(TAG,
                            "do_GET_active_tasks Transition [" + event.getTransition().getTrigger() +
                                    "] Source:" + event.getTransition().getSource() +
                                    ", Destination:" + event.getTransition().getDestination());
                }

                if (longpoll) {
                    Log.w(TAG, "Router: Sending longpoll replication response");
                    sendResponse();
                    if (callbackBlock != null) {
                        byte[] data = null;
                        try {
                            data = Manager.getObjectMapper().writeValueAsBytes(activity);
                        } catch (Exception e) {
                            Log.w(TAG, "Error serializing JSON", e);
                        }
                        OutputStream os = connection.getResponseOutputStream();
                        try {
                            os.write(data);
                            os.close();
                        } catch (IOException e) {
                            Log.e(TAG, "IOException writing to internal streams", e);
                        }
                    }
                    //remove this change listener because a new one will be added when this responds
                    event.getSource().removeChangeListener(self);
                } else {
                    Log.w(TAG, "Router: Sending continous replication change chunk");
                    sendContinuousReplicationChanges(activity);
                }
            }
        };

        List<Map<String, Object>> activities = new ArrayList<Map<String, Object>>();
        for (Database db : manager.allOpenDatabases()) {
            List<Replication> activeReplicators = db.getAllReplications();
            if (activeReplicators != null) {
                for (Replication replicator : activeReplicators) {
                    if (replicator.isRunning()) {
                        Map<String, Object> activity = getActivity(replicator);

                        if (session_id != null) {
                            if (replicator.getSessionID().equals(session_id)) {
                                activities.add(activity);
                            }
                        } else {
                            activities.add(activity);
                        }

                        if (continuous || longpoll) {
                            if (session_id != null) {
                                if (replicator.getSessionID().equals(session_id)) {
                                    replicator.addChangeListener(listener);
                                }
                            } else {
                                replicator.addChangeListener(listener);
                            }
                        }
                    }
                }
            }
        }

        if (continuous || longpoll) {
            connection.setChunked(true);
            connection.setResponseCode(Status.OK);
            sendResponse();
            if (continuous && !activities.isEmpty()) {
                for (Map<String, Object> activity : activities) {
                    sendContinuousReplicationChanges(activity);
                }
            }

            // Don't close connection; more data to come
            return new Status(0);
        } else {
            connection.setResponseBody(new Body(activities));
            return new Status(Status.OK);
        }
    }
561284
function (context, callback) {

            var self = this,
                options = {},
                model,
                processor,
                isPatch = context.request.method === "PATCH",
                upsert = this.$resourceConfiguration.$.upsert,
                defaults;


            flow()
                .seq("collection", function (cb) {
                    self._findCollection(context, cb);
                })
                .seq(function (cb) {
                    model = this.vars.collection.createItem(self.$resourceId);
                    defaults = _.clone(model.$);
                    // check if model exists
                    if (upsert && !isPatch) {
                        cb();
                    } else {
                        model.fetch(null, cb);
                    }
                })
                .seq(function () {
                    if (!isPatch) {
                        model.clear();
                        model.set(defaults);
                    }

                    var payload = context.request.params;

                    processor = self.$restHandler.$restDataSource.getProcessorForModel(model);

                    model.set(processor.parse(model, payload || {}));

                    options.action = DataSource.ACTION.UPDATE;

                    // TODO: add hook to add session data like user id
                    if (upsert === true) {
                        options.upsert = true;
                        model.set(model.idField, model.factory.prototype.convertIdentifier(self.$resourceId));
                    }
                })
                .seq(function () {
                    self._generateAutoValues(model);
                })
                .seq(function (cb) {
                    self._validate(model, context, cb);
                })
                .seq(function (cb) {
                    self._beforeModelSave(model, context, cb);
                })
                .seq(function (cb) {
                    self._beforeModelUpdate(model, context, cb);
                })
                .seq(function (cb) {
                    model.validateAndSave(options, cb);
                })
                .seq(function (cb) {
                    self._afterModelUpdate(model, context, cb);
                })
                .exec(function (err, results) {
                    if (!err) {
                        // TODO: do correct invalidation
                        results.collection.invalidatePageCache();

                        self._fetchAllHrefsForModel(model, context);

                        // TODO: generate the location header
                        var body = JSON.stringify(processor.compose(model, null));

                        var response = context.response;
                        response.writeHead(200, "", {
                            'Content-Type': 'application/json'
                            // TODO : add updated date in head ???
//                            'Location': context.request.urlInfo.uri
                        });

                        response.write(body);
                        response.end();

                        callback(null);
                    } else {
                        if (!(err instanceof HttpError)) {
                            var statusCode = 400;
                            if (err === DataSource.ERROR.NOT_FOUND) {
                                statusCode = 404;
                            }
                            err = new HttpError(err, statusCode);
                        }
                        callback(err);
                    }
                });

        }
546024
function getUnchangedTasks(globs) {
  var prevPaths;
  var nextPaths;
  var unchangedPaths;
  var prevFiles;
  var nextFiles;
  var tasks1;
  var paths1;
  var tasks;
  var paths;

  // time: current
  return getFlatUniquePaths(globs)
    .then(function(res) {
      nextPaths = res;
      return gitCmds.safeStashAll();
    })
    .then(function() {
      // time: prev
      return getFlatUniquePaths(globs);
    })
    .then(function(res) {
      prevPaths = res;
      unchangedPaths = _.intersection(prevPaths, nextPaths);

      if (unchangedPaths.length === 0) {
        return Promise.reject(
          new Error('No Huxleyfile found that is unchanged.')
        );
      }

      // at this point, we got all the `Huxleyfile.json`s directory paths that
      // exist before and after stashing
      return Promise.map(unchangedPaths, loadJSON);
    })
    .then(function(res) {
      prevFiles = res;
      return gitCmds.safeUnstashAll();
    })
    .then(function() {
      // time: current
      return Promise.map(unchangedPaths, loadJSON);
    })
    .then(function(res) {
      // at this point, we read all the `Huxleyfile.json`s of those paths
      nextFiles = res;

      var unchangedContainerPaths = unchangedPaths.map(path.dirname);
      var res1 = filterFilesForUnchangedTasks(
        prevFiles,
        nextFiles,
        unchangedContainerPaths
      );

      // not done! We need to make sure these tasks' `record.json`s actually
      // exist
      return filterExistingRecords(res1[0], res1[1]);
    })
    .spread(function(a, b) {
      tasks1 = a;
      paths1 = b;
      return gitCmds.safeStashAll();
    })
    .then(function() {
      // time: prev
      return filterExistingRecords(tasks1, paths1);
    })
    .spread(function(a, b) {
      tasks = a;
      paths = b;
      return gitCmds.safeUnstashAll();
    })
    .then(function() {
      // time: current
      if (tasks.length === 0) {
        return Promise.reject(new Error('No task found that is unchanged.'));
      }

      return [tasks, paths];
    });
}
57929
function moduleTask(taskname, depsOrFn, fn, noAutoRegister) {

    var module = this;

    if (typeof(depsOrFn) == 'string') {
      depsOrFn = [depsOrFn];
    }

    if (Array.isArray(depsOrFn)) {

      var map = [];

      depsOrFn.forEach(function (dep) {
        map.push(module.name + '-' + dep);
      });

      depsOrFn = map;

    }

    var fulltaskname = module.name + '-' + taskname;
    gulp.task(fulltaskname, depsOrFn, fn);

    if(!noAutoRegister) {
      module.tasks.push(fulltaskname);
    }

  }
572130
public void doUpdates(MetricsContext unused) {
    // In case of running in LocalMode tracker == null
    if (tracker != null) {
      synchronized (tracker) {
        synchronized (this) {
          numRunningMaps = 0;
          numRunningReduces = 0;

          numWaitingMaps = 0;
          numWaitingReduces = 0;
          numTasksInMemory = 0;

          List<JobInProgress> jobs = tracker.getRunningJobs();
          for (JobInProgress jip : jobs) {
            for (TaskInProgress tip : jip.maps) {
              if (tip.isRunning()) {
                numRunningMaps++;
              } else if (tip.isRunnable()) {
                numWaitingMaps++;
              }
            }
            for (TaskInProgress tip : jip.reduces) {
              if (tip.isRunning()) {
                numRunningReduces++;
              } else if (tip.isRunnable()) {
                numWaitingReduces++;
              }

            }
            numTasksInMemory += jip.getTasks(TaskType.MAP).length;
            numTasksInMemory += jip.getTasks(TaskType.REDUCE).length;
          }

          // Get tracker metrics
          numTrackersDead = tracker.getDeadNodes().size();
          ClusterStatus cs = tracker.getClusterStatus(false);
          numTrackersExcluded = cs.getNumExcludedNodes();
        }
      }
    }
    synchronized (this) {
      metricsRecord.setMetric("map_slots", numMapSlots);
      metricsRecord.setMetric("reduce_slots", numReduceSlots);
      metricsRecord.incrMetric("blacklisted_maps", numBlackListedMapSlots);
      metricsRecord.incrMetric("blacklisted_reduces",
          numBlackListedReduceSlots);
      metricsRecord.incrMetric("jobs_submitted", numJobsSubmitted);
      metricsRecord.incrMetric("jobs_completed", numJobsCompleted);
      metricsRecord.setMetric("waiting_maps", numWaitingMaps);
      metricsRecord.setMetric("waiting_reduces", numWaitingReduces);

      metricsRecord.incrMetric("reserved_map_slots", numReservedMapSlots);
      metricsRecord.incrMetric("reserved_reduce_slots", numReservedReduceSlots);
      metricsRecord.incrMetric("occupied_map_slots", numOccupiedMapSlots);
      metricsRecord.incrMetric("occupied_reduce_slots", numOccupiedReduceSlots);

      metricsRecord.incrMetric("jobs_failed", numJobsFailed);
      metricsRecord.incrMetric("jobs_killed", numJobsKilled);

      metricsRecord.incrMetric("jobs_preparing", numJobsPreparing);
      metricsRecord.incrMetric("jobs_running", numJobsRunning);

      metricsRecord.setMetric("running_maps", numRunningMaps);
      metricsRecord.setMetric("running_reduces", numRunningReduces);

      metricsRecord.setMetric("num_tasks_in_memory", numTasksInMemory);

      metricsRecord.setMetric("trackers", numTrackers);
      metricsRecord.setMetric("trackers_blacklisted", numTrackersBlackListed);
      metricsRecord.setMetric("trackers_decommissioned",
          numTrackersDecommissioned);
      metricsRecord.setMetric("trackers_excluded", numTrackersExcluded);
      metricsRecord.setMetric("trackers_dead", numTrackersDead);

      metricsRecord.incrMetric("num_launched_jobs", numJobsLaunched);
      metricsRecord.incrMetric("total_submit_time", totalSubmitTime);

      aggregateJobStats.incrementMetricsAndReset(metricsRecord);

      // Update additional speculation stats for measuring the performance
      // of different kinds of speculation
      for (SpecStats.TaskType taskType : SpecStats.TaskType.values()) {
        for (SpecStats.SpecType specType : SpecStats.SpecType.values()) {
          for(SpecStats.StatType statType : SpecStats.StatType.values()) {
            String key = "speculation_by_" + specType.toString().toLowerCase()  
                + "_" + "rate_" + taskType.toString().toLowerCase() + "_" + 
                statType.toString().toLowerCase();
            long value = specStats.getStat(taskType, specType, statType);
            metricsRecord.setMetric(key, value);
          }
        }
      }

      
      for (Group group: countersToMetrics) {
        String groupName = group.getName();
        for (Counter counter : group) {
          String name = groupName + "_" + counter.getName();
          name = name.replaceAll("[^a-zA-Z_]", "_").toLowerCase();
          metricsRecord.incrMetric(name, counter.getValue());
        }
      }
      clearCounters();

      numJobsSubmitted = 0;
      numJobsCompleted = 0;
      numWaitingMaps = 0;
      numWaitingReduces = 0;
      numBlackListedMapSlots = 0;
      numBlackListedReduceSlots = 0;

      numReservedMapSlots = 0;
      numReservedReduceSlots = 0;
      numOccupiedMapSlots = 0;
      numOccupiedReduceSlots = 0;

      numJobsFailed = 0;
      numJobsKilled = 0;

      numJobsPreparing = 0;
      numJobsRunning = 0;

      numRunningMaps = 0;
      numRunningReduces = 0;

      numTrackers = 0;
      numTrackersBlackListed = 0;

      totalSubmitTime = 0;
      numJobsLaunched = 0;
    }
    metricsRecord.update();
  }
1273447
async function _handleChangeAndPush(event, doc, timestamp) {
  // Update the resource content and set dirty
  // TODO: Remove one of these steps since it does encryption twice
  // in the case where the resource does not exist yet
  const resource = await getOrCreateResourceForDoc(doc);

  const updatedResource = await store.updateResource(resource, {
    name: doc.name || 'n/a',
    lastEdited: timestamp,
    lastEditedBy: session.getAccountId(),
    encContent: await encryptDoc(resource.resourceGroupId, doc),
    removed: event === db.CHANGE_REMOVE,
    dirty: true,
  });

  // Debounce pushing of dirty resources
  logger.debug(`Queue ${event} ${updatedResource.id}`);
}
1309827
function updateTask(task, callback) {
    database.serialize(function() {
      var expiry_time = getExpiryTime(task);
      var update_task = "UPDATE "+table_name+" SET task = ?, "
      +"at = datetime("+expiry_time+") WHERE "+table_name+".id = ?";

      log.debug('sqlite-adapter','Running query:',update_task);
      database.run(update_task, JSON.stringify(task), task.id, function(error) {
        if (error) {
          callback(error);
        }
        callback(error, this.changes);
      });
    });
  }
823736
private int processCommits(GitHubRepo repo) {
        int count = 0;
        Long existingCount = commitRepository.countCommitsByCollectorItemId(repo.getId());
        if (existingCount == 0) {
            List<Commit> newCommits = gitHubClient.getCommits();
            newCommits.forEach(c -> c.setCollectorItemId(repo.getId()));
            Iterable<Commit> saved = commitRepository.save(newCommits);
            count = saved != null ? Lists.newArrayList(saved).size() : 0;
        } else {
            Collection<Commit> nonDupCommits = gitHubClient.getCommits().stream()
                    .<Map<String, Commit>> collect(HashMap::new,(m,c)->m.put(c.getScmRevisionNumber(), c), Map::putAll)
                    .values();
            for (Commit commit : nonDupCommits) {
                LOG.debug(commit.getTimestamp() + ":::" + commit.getScmCommitLog());
                if (isNewCommit(repo, commit)) {
                    commit.setCollectorItemId(repo.getId());
                    commitRepository.save(commit);
                    count++;
                }
            }
        }
        LOG.info("-- Saved Commits = " + count);
        return count;
    }
963996
def _insert_tasks(tasks, queue, transactional=False,
                  retry_transient_errors=True,
                  retry_delay=RETRY_SLEEP_SECS):
    
    from google.appengine.api import taskqueue

    if not tasks:
        return 0

    try:
        taskqueue.Queue(name=queue).add(tasks, transactional=transactional)
        return len(tasks)
    except (taskqueue.BadTaskStateError,
            taskqueue.TaskAlreadyExistsError,
            taskqueue.TombstonedTaskError):
        if len(tasks) <= 1:
            # Task has already been inserted, no reason to report an error here.
            return 0

        # If a list of more than one Tasks is given, a raised exception does
        # not guarantee that no tasks were added to the queue (unless
        # transactional is set to True). To determine which tasks were
        # successfully added when an exception is raised, check the
        # Task.was_enqueued property.
        reinsert = _tasks_to_reinsert(tasks, transactional)
        count = len(reinsert)
        inserted = len(tasks) - count
        inserted += _insert_tasks(reinsert[:count / 2], queue, transactional,
                                  retry_transient_errors, retry_delay)
        inserted += _insert_tasks(reinsert[count / 2:], queue, transactional,
                                  retry_transient_errors, retry_delay)

        return inserted
    except taskqueue.TransientError:
        # Always re-raise for transactional insert, or if specified by
        # options.
        if transactional or not retry_transient_errors:
            raise

        reinsert = _tasks_to_reinsert(tasks, transactional)

        # Retry with a delay, and then let any errors re-raise.
        time.sleep(retry_delay)

        taskqueue.Queue(name=queue).add(reinsert, transactional=transactional)
        return len(tasks)
289922
def run_on_changes(modified, added, removed)
      types = {
        MODIFICATION_TASKS => modified,
        ADDITION_TASKS => added,
        REMOVAL_TASKS => removed
      }

      UI.clearable

      Guard.state.scope.grouped_plugins.each do |_group, plugins|
        _run_group_plugins(plugins) do |plugin|
          UI.clear
          types.each do |tasks, unmatched_paths|
            next if unmatched_paths.empty?
            match_result = Watcher.match_files(plugin, unmatched_paths)
            next if match_result.empty?
            task = tasks.detect { |meth| plugin.respond_to?(meth) }
            _supervise(plugin, task, match_result) if task
          end
        end
      end
    end