<commit_before>#! /usr/bin/env python

import os

os.environ['DJANGO_SETTINGS_MODULE'] = 'astrometry.net.settings'

import sys
import socket
import time
import tempfile
import thread
from datetime import datetime
from optparse import OptionParser

from urllib import urlencode
from urllib2 import urlopen

from django.db import transaction

import pyfits

import astrometry.net.settings as settings

#from astrometry.net.server.log import log, reconfig_logging

hostname = socket.gethostname()
hostname = hostname.split('.')[0]
pid = os.getpid()
settings.SERVER_LOGFILE = settings.LOG_DIR + 'worker-%s-%i.log' % (hostname, pid)
print 'Logging to', settings.SERVER_LOGFILE

from astrometry.net.server.log import log

from astrometry.net.server.models import *
from astrometry.util.run_command import run_command
from astrometry.net.portal.wcs import TanWCS
from astrometry.net.portal.job import Calibration, Job

def get_header(header, key, default):
    try:
        return header[key]
    except KeyError:
        return default

#def log(*args):
#    pass

class Solver(object):
    def __init__(self, q, indexdirs):
        self.worker = Worker(queue=q)
        self.q = q
        self.worker.save()
        self.worker.start_keepalive_thread()
        self.aborting_job = False

        indexpaths = []

        for d in indexdirs:
            if not os.path.exists(d):
                print 'No such directory:', d
                continue
            files = os.listdir(d)
            for f in files:
                csuff = '.ckdt.fits'
                qsuff = '.quad.fits'
                ssuff = '.skdt.fits'
                if not f.endswith(qsuff):
                    continue
                base = os.path.join(d, f[:-len(qsuff)])
                if not (os.path.exists(base + csuff) and
                        os.path.exists(base + ssuff)):
                    continue
                print 'Found index', base
                qfile = base + qsuff
                hdus = pyfits.open(qfile)
                hdr = hdus[0].header
                indexid = get_header(hdr, 'INDEXID', None)
                hp = get_header(hdr, 'HEALPIX', -1)
                hpnside = get_header(hdr, 'HPNSIDE', 1)
                scalelo = float(get_header(hdr, 'SCALE_L', 0))
                scalehi = float(get_header(hdr, 'SCALE_U', 0))
                print 'id', indexid, 'hp', hp, 'hp nside', hpnside, 'scale [%g, %g]' % (scalelo, scalehi)
                if indexid is None:
                    continue

                (ind, nil) = Index.objects.get_or_create(indexid=indexid,
                                                         healpix=hp,
                                                         healpix_nside=hpnside,
                                                         defaults={'scalelo': scalelo,
                                                                   'scalehi': scalehi,})
                self.worker.indexes.add(ind)
                self.worker.save()
                indexpaths.append(base)

        print 'My indexes:', self.worker.pretty_index_list()

        # Write my backend.cfg file.
        (f, self.backendcfg) = tempfile.mkstemp('', 'backend.cfg-')
        os.close(f)
        f = open(self.backendcfg, 'wb')
        f.write('\n'.join(['inparallel'] +
                          ['index %s' % path for path in indexpaths]
                          ))
        f.close()

    #<EMAIL>
    def claim_work(self):
        #transaction.commit()

        # pull candidate jobs from the queue
        qjobs = (QueuedJob.objects.all()
                 .filter(q=q, done=False)
                 .filter(work__inprogress=False, work__done=False)
                 .filter(work__index__in=[self.indexes.all()])
                 .order_by('priority', 'enqueuetime'))
        if len(qjobs) == 0:
            return (None,None)

        myinds = set(str(i) for i in self.indexes.all())
        for j in qjobs:
            requested = set(str(w.index) for w in
                            j.work.all()
                            .filter(inprogress=False, done=False))
            incommon = requested.intersection(myinds)
            #print
            #print 'QJob', j
            #print 'Requested work:', requested
            #print 'My indexes:', myinds
            #print 'Incommon:', incommon
            #print
            if len(incommon) == 0:
                continue

            # test and set
            mywork = [w for w in j.work.all() if str(w.index) in incommon]
            for w in mywork:
                print '  index', w.index

                # UPDATE SET worker=me, inprogress=true WHERE inprogress=false
                # ?

                #w.worker = self.worker
                #w.inprogress = True
                #w.save()

            # pull out the list of Work items I successfully claimed.
            mywork = mywork.filter(worker=self.worker)
            return (j, mywork)

        return (None, None)




    def run_one(self):
        nextwork = self.worker.get_next_work(self.q)
        if nextwork is None:
            return False
        (qjob, work) = nextwork

        print 'Got my next job:', qjob, 'work', work

        for w in work:
            print '  index', w.index
            w.worker = self.worker
            w.inprogress = True
            w.save()

        self.worker.job = qjob
        self.worker.save()<commit_msg>try locking...<commit_after>#! /usr/bin/env python

import os

os.environ['DJANGO_SETTINGS_MODULE'] = 'astrometry.net.settings'

import sys
import socket
import time
import tempfile
import thread
from datetime import datetime
from optparse import OptionParser

from urllib import urlencode
from urllib2 import urlopen

from django.db import transaction
from django.db import connection

import pyfits

import astrometry.net.settings as settings

#from astrometry.net.server.log import log, reconfig_logging

hostname = socket.gethostname()
hostname = hostname.split('.')[0]
pid = os.getpid()
settings.SERVER_LOGFILE = settings.LOG_DIR + 'worker-%s-%i.log' % (hostname, pid)
print 'Logging to', settings.SERVER_LOGFILE

from astrometry.net.server.log import log

from astrometry.net.server.models import *
from astrometry.util.run_command import run_command
from astrometry.net.portal.wcs import TanWCS
from astrometry.net.portal.job import Calibration, Job

def get_header(header, key, default):
    try:
        return header[key]
    except KeyError:
        return default

#def log(*args):
#    pass

class Solver(object):
    def __init__(self, q, indexdirs):
        self.worker = Worker(queue=q)
        self.q = q
        self.worker.save()
        self.worker.start_keepalive_thread()
        self.aborting_job = False

        indexpaths = []

        for d in indexdirs:
            if not os.path.exists(d):
                print 'No such directory:', d
                continue
            files = os.listdir(d)
            for f in files:
                csuff = '.ckdt.fits'
                qsuff = '.quad.fits'
                ssuff = '.skdt.fits'
                if not f.endswith(qsuff):
                    continue
                base = os.path.join(d, f[:-len(qsuff)])
                if not (os.path.exists(base + csuff) and
                        os.path.exists(base + ssuff)):
                    continue
                print 'Found index', base
                qfile = base + qsuff
                hdus = pyfits.open(qfile)
                hdr = hdus[0].header
                indexid = get_header(hdr, 'INDEXID', None)
                hp = get_header(hdr, 'HEALPIX', -1)
                hpnside = get_header(hdr, 'HPNSIDE', 1)
                scalelo = float(get_header(hdr, 'SCALE_L', 0))
                scalehi = float(get_header(hdr, 'SCALE_U', 0))
                print 'id', indexid, 'hp', hp, 'hp nside', hpnside, 'scale [%g, %g]' % (scalelo, scalehi)
                if indexid is None:
                    continue

                (ind, nil) = Index.objects.get_or_create(indexid=indexid,
                                                         healpix=hp,
                                                         healpix_nside=hpnside,
                                                         defaults={'scalelo': scalelo,
                                                                   'scalehi': scalehi,})
                self.worker.indexes.add(ind)
                self.worker.save()
                indexpaths.append(base)

        print 'My indexes:', self.worker.pretty_index_list()

        # Write my backend.cfg file.
        (f, self.backendcfg) = tempfile.mkstemp('', 'backend.cfg-')
        os.close(f)
        f = open(self.backendcfg, 'wb')
        f.write('\n'.join(['inparallel'] +
                          ['index %s' % path for path in indexpaths]
                          ))
        f.close()

    @transaction.commit_manually
    def claim_work(self):

        cursor = connection.cursor()
        cursor.execute('LOCK TABLE %s IN SHARE ROW EXCLUSIVE MODE' %
                       QueuedJob._meta.db_table)

        # pull candidate jobs from the queue
        qjobs = (QueuedJob.objects.all()
                 .filter(q=q, done=False)
                 .filter(work__inprogress=False, work__done=False)
                 .filter(work__index__in=[self.indexes.all()])
                 .order_by('priority', 'enqueuetime'))
        if qjobs.count() == 0:
            return (None,None)

        myinds = set(str(i) for i in self.indexes.all())
        for j in qjobs:
            requested = set(str(w.index) for w in
                            j.work.all()
                            .filter(inprogress=False, done=False))
            incommon = requested.intersection(myinds)
            if len(incommon) == 0:
                continue

            mywork = [w for w in j.work.all() if str(w.index) in incommon]
            for w in mywork:
                w.worker = self.worker
                w.inprogress = True
                w.save()

            transaction.commit()
            return (j, mywork)

        return (None, None)



    def run_one(self):
        #nextwork = self.worker.get_next_work(self.q)
        #if nextwork is None:
        #    return False
        #(qjob, work) = nextwork
        #print 'Got my next job:', qjob, 'work', work
        #for w in work:
        #    print '  index', w.index
        #    w.worker = self.worker
        #    w.inprogress = True
        #    w.save()

        (qjob, work) = self.claim_work()
        if qjob is None:
            return False

        self.worker.job = qjob
        self.worker.save()
{'fraction_non_alphanumeric': 0.10335431739171286, 'fraction_numerical': 0.0016912524664098468, 'mean_word_length': 1.5464114832535885, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5483469', 'n_tokens_mistral': 3141, 'n_tokens_neox': 3014, 'n_words': 758}