<commit_before>package org.neo4j.util;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.neo4j.api.core.NeoService;
import org.neo4j.api.core.Node;
import org.neo4j.api.core.Transaction;

public abstract class NeoQueueWorker extends Thread
{
    private final NeoService neo;
    private final NeoQueue queue;
    private boolean halted;
    private boolean requestedToPause;
    private boolean paused;
    private int batchSize;
    
    public NeoQueueWorker( NeoService neo, NeoQueue queue, int batchSize )
    {
        super( "NeoQueueWorker" );
        this.neo = neo;
        this.queue = queue;
        this.batchSize = batchSize;
    }
    
    public NeoQueue getQueue()
    {
        return this.queue;
    }
    
    public void setPaused( boolean paused )
    {
        if ( this.paused == paused )
        {
            return;
        }
        
        if ( paused && this.requestedToPause )
        {
            waitUntilReallyPaused();
            return;
        }
        
        this.requestedToPause = paused;
        if ( paused )
        {
            waitUntilReallyPaused();
        }
        else
        {
            this.paused = false;
        }
    }
    
    private void waitUntilReallyPaused()
    {
        while ( !this.paused )
        {
            try
            {
                Thread.sleep( 100 );
            }
            catch ( InterruptedException e )
            { // OK
            }
        }
    }

    public boolean isPaused()
    {
        return this.paused;
    }
    
    private void sleepQuiet( long millis )
    {
        try
        {
            Thread.sleep( millis );
        }
        catch ( InterruptedException e )
        {
            // Ok
        }
    }
    
    @Override
    public void run()
    {
        while ( !this.halted )
        {
            if ( this.requestedToPause || this.paused )
            {
                this.paused = true;
                this.requestedToPause = false;
                sleepQuiet( 1000 );
                continue;
            }
            
            if ( !executeOneBatch() )
            {
                sleepQuiet( 100 );
            }
        }
    }
    
    public void add( Map<String, Object> values )
    {
        Node entry = this.queue.add();
        for ( Map.Entry<String, Object> value : values.entrySet() )
        {
            entry.setProperty( value.getKey(), value.getValue() );
        }
    }
    
    protected void beforeBatch()
    {
    }
    
    protected void afterBatch()
    {
    }
    
    private boolean executeOneBatch()
    {
        int entrySize = 0;
        Collection<Map<String, Object>> entries = null;
        Transaction tx = neo.beginTx();
        try
        {
            Node[] nodes = this.queue.peek( batchSize );
            if ( nodes.length == 0 )
            {
                return false;
            }
            entrySize = nodes.length;
            
            entries = new ArrayList<Map<String,Object>>( entrySize );
            for ( Node node : nodes )
            {
                entries.add( readNode( node ) );
            }
            tx.success();
        }
        finally
        {
            tx.finish();
        }
        
        beforeBatch();
        for ( Map<String, Object> entry : entries )
        {
            doOne( entry );
        }
        afterBatch();
        final int size = entrySize;
        new DeadlockCapsule<Object>( "remover" )
        {
            @Override
            public Object tryOnce()
            {
                queue.remove( size );
                return null;
            }
        }.run();
        return true;
    }
    
    private Map<String, Object> readNode( Node node )
    {
        Map<String, Object> result = new HashMap<String, Object>();
        for ( String key : node.getPropertyKeys() )
        {
            result.put( key, node.getProperty( key ) );
        }
        return result;
    }

    private void doOne( Map<String, Object> entry )
    {
        // Try a max of ten times if it fails.
        Exception exception = null;
        for ( int i = 0; !this.halted && i < 10; i++ )
        {
            try
            {
                doHandleEntry( entry );
                return;
            }
            catch ( Exception e )
            {
                exception = e;
            }
            sleepQuiet( 500 );
        }<commit_msg>Wraps the entire batch in one tx... is this good or not?<commit_after>package org.neo4j.util;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import org.neo4j.api.core.NeoService;
import org.neo4j.api.core.Node;
import org.neo4j.api.core.Transaction;

public abstract class NeoQueueWorker extends Thread
{
    private final NeoService neo;
    private final NeoQueue queue;
    private boolean halted;
    private boolean requestedToPause;
    private boolean paused;
    private int batchSize;
    
    public NeoQueueWorker( NeoService neo, NeoQueue queue, int batchSize )
    {
        super( "NeoQueueWorker" );
        this.neo = neo;
        this.queue = queue;
        this.batchSize = batchSize;
    }
    
    public NeoQueue getQueue()
    {
        return this.queue;
    }
    
    public void setPaused( boolean paused )
    {
        if ( this.paused == paused )
        {
            return;
        }
        
        if ( paused && this.requestedToPause )
        {
            waitUntilReallyPaused();
            return;
        }
        
        this.requestedToPause = paused;
        if ( paused )
        {
            waitUntilReallyPaused();
        }
        else
        {
            this.paused = false;
        }
    }
    
    private void waitUntilReallyPaused()
    {
        while ( !this.paused )
        {
            try
            {
                Thread.sleep( 100 );
            }
            catch ( InterruptedException e )
            { // OK
            }
        }
    }

    public boolean isPaused()
    {
        return this.paused;
    }
    
    private void sleepQuiet( long millis )
    {
        try
        {
            Thread.sleep( millis );
        }
        catch ( InterruptedException e )
        {
            // Ok
        }
    }
    
    @Override
    public void run()
    {
        while ( !this.halted )
        {
            if ( this.requestedToPause || this.paused )
            {
                this.paused = true;
                this.requestedToPause = false;
                sleepQuiet( 1000 );
                continue;
            }
            
            if ( !executeOneBatch() )
            {
                sleepQuiet( 100 );
            }
        }
    }
    
    public void add( Map<String, Object> values )
    {
        Node entry = this.queue.add();
        for ( Map.Entry<String, Object> value : values.entrySet() )
        {
            entry.setProperty( value.getKey(), value.getValue() );
        }
    }
    
    protected void beforeBatch()
    {
    }
    
    protected void afterBatch()
    {
    }
    
    private boolean executeOneBatch()
    {
        int entrySize = 0;
        Collection<Map<String, Object>> entries = null;
        Transaction tx = neo.beginTx();
        try
        {
            Node[] nodes = this.queue.peek( batchSize );
            if ( nodes.length == 0 )
            {
                return false;
            }
            entrySize = nodes.length;
            
            entries = new ArrayList<Map<String,Object>>( entrySize );
            for ( Node node : nodes )
            {
                entries.add( readNode( node ) );
            }

            beforeBatch();
            try
            {
                for ( Map<String, Object> entry : entries )
                {
                    doOne( entry );
                }
            }
            finally
            {
                afterBatch();
            }
            
            final int size = entrySize;
            new DeadlockCapsule<Object>( "remover" )
            {
                @Override
                public Object tryOnce()
                {
                    queue.remove( size );
                    return null;
                }
            }.run();
            
            tx.success();
        }
        finally
        {
            tx.finish();
        }
        return true;
    }
    
    private Map<String, Object> readNode( Node node )
    {
        Map<String, Object> result = new HashMap<String, Object>();
        for ( String key : node.getPropertyKeys() )
        {
            result.put( key, node.getProperty( key ) );
        }
        return result;
    }

    private void doOne( Map<String, Object> entry )
    {
        // Try a max of ten times if it fails.
        Exception exception = null;
        for ( int i = 0; !this.halted && i < 10; i++ )
        {
            try
            {
                doHandleEntry( entry );
                return;
            }
            catch ( Exception e )
            {
                exception = e;
            }
            sleepQuiet( 500 );
        }
{'fraction_non_alphanumeric': 0.09127458437466043, 'fraction_numerical': 0.004781049657720309, 'mean_word_length': 1.1439552760307476, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1453839', 'n_tokens_mistral': 2569, 'n_tokens_neox': 2449, 'n_words': 577}