<commit_before>/**
 *  Caretaker
 *  Copyright 22.02.2015 by <NAME>, @0rb1t3r
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *  
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Lesser General Public License for more details.
 *  
 *  You should have received a copy of the GNU Lesser General Public License
 *  along with this program in the file lgpl21.txt
 *  If not, see <http://www.gnu.org/licenses/>.
 */

package org.loklak;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.elasticsearch.search.sort.SortOrder;
import org.loklak.api.p2p.HelloService;
import org.loklak.api.p2p.PushServlet;
import org.loklak.api.search.SuggestServlet;
import org.loklak.data.DAO;
import org.loklak.data.DAO.IndexName;
import org.loklak.data.IncomingMessageBuffer;
import org.loklak.harvester.TwitterAPI;
import org.loklak.objects.QueryEntry;
import org.loklak.objects.TwitterTimeline;
import org.loklak.objects.BasicTimeline.Order;
import org.loklak.tools.DateParser;
import org.loklak.tools.OS;

import twitter4j.TwitterException;

/**
 * The caretaker class is a concurrent thread which does peer-to-peer operations
 * and data transmission asynchronously.
 */
public class Caretaker extends Thread {

    private static final Random random = new Random(System.currentTimeMillis());
    
    private boolean shallRun = true;
    
    public  final static long startupTime = System.currentTimeMillis();
    private final static long upgradeWait = DateParser.DAY_MILLIS; // 1 day
    public        static long upgradeTime = startupTime + upgradeWait;
    private final static long helloPeriod = 600000; // one ping each 10 minutes
    private       static long helloTime   = 0; // latest hello ping time
    private       static long deletionTime = 0;

    private static final int TIMELINE_PUSH_MINSIZE = 200;
    private static final int TIMELINE_PUSH_MAXSIZE = 1000;
    
    /**
     * ask the thread to shut down
     */
    public void shutdown() {
        this.shallRun = false;
        this.interrupt();
        DAO.log("catched caretaker termination signal");
    }
    
    @Override
    public void run() {
        Thread.currentThread().setName("CARETAKER");
        // send a message to other peers that I am alive
        String[] backends = DAO.getBackend();
        int maxRetries = Integer.valueOf(DAO.getConfig("caretaker.backendpush.retries", "5"));
        int backOffParameter = Integer.valueOf(DAO.getConfig("caretaker.backendpush.backoff", "3000"));

        boolean busy = false;
        // work loop
        beat: while (this.shallRun) try {
            // check upgrade time
            if (System.currentTimeMillis() > upgradeTime) {
                // increase the upgrade time to prevent that the peer runs amok (re-tries the attempt all the time) when upgrade fails for any reason
                upgradeTime = upgradeTime + upgradeWait;
                
                // do an upgrade
                DAO.log("UPGRADE: starting an upgrade");
                upgrade();
                DAO.log("UPGRADE: started an upgrade");
            }
            
            // check ping
            if (System.currentTimeMillis() - helloPeriod > helloTime) {
                helloTime = System.currentTimeMillis();
                HelloService.propagate(backends);
            }
            
            // clear caches
            if (SuggestServlet.cache.size() > 100) SuggestServlet.cache.clear();
            
            // sleep a bit to prevent that the DoS limit fires at backend server
            try {Thread.sleep(busy ? 500 : 5000);} catch (InterruptedException e) {}
            if (!this.shallRun) break beat;
            busy = false;
            
            //DAO.log("connection pool: " + ClientConnection.cm.getTotalStats().toString());
            
            // peer-to-peer operation
            TwitterTimeline tl = DAO.outgoingMessages.takeTimelineMin(Order.CREATED_AT, TIMELINE_PUSH_MINSIZE, TIMELINE_PUSH_MAXSIZE);
            int tl_size = tl == null ? 0 : tl.size();
            if (tl_size > 0 && backends.length > 0) {
                // transmit the timeline
                long start = System.currentTimeMillis();
                boolean success = PushServlet.push(backends, tl);
                if (success) {
                    DAO.log("success pushing " + tl.size() + " messages to backend " + Arrays.toString(backends) + " in 1st attempt in " + (System.currentTimeMillis() - start) + " ms");
                } else {
                    // we should try again.. but not an infinite number because then
                    // our timeline in RAM would fill up our RAM creating a memory leak<commit_msg>missed<commit_after>/**
 *  Caretaker
 *  Copyright 22.02.2015 by <NAME>, @0rb1t3r
 *
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *  
 *  This library is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Lesser General Public License for more details.
 *  
 *  You should have received a copy of the GNU Lesser General Public License
 *  along with this program in the file lgpl21.txt
 *  If not, see <http://www.gnu.org/licenses/>.
 */

package org.loklak;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.elasticsearch.search.sort.SortOrder;
import org.loklak.api.p2p.HelloService;
import org.loklak.api.p2p.PushServlet;
import org.loklak.api.search.SuggestServlet;
import org.loklak.data.DAO;
import org.loklak.data.DAO.IndexName;
import org.loklak.data.IncomingMessageBuffer;
import org.loklak.harvester.TwitterAPI;
import org.loklak.objects.QueryEntry;
import org.loklak.objects.TwitterTimeline;
import org.loklak.objects.BasicTimeline.Order;
import org.loklak.tools.DateParser;
import org.loklak.tools.OS;

import twitter4j.TwitterException;

/**
 * The caretaker class is a concurrent thread which does peer-to-peer operations
 * and data transmission asynchronously.
 */
public class Caretaker extends Thread {

    private static final Random random = new Random(System.currentTimeMillis());
    
    private boolean shallRun = true;
    
    public  final static long startupTime = System.currentTimeMillis();
    private final static long upgradeWait = DateParser.DAY_MILLIS; // 1 day
    public        static long upgradeTime = startupTime + upgradeWait;
    private final static long helloPeriod = 600000; // one ping each 10 minutes
    private       static long helloTime   = 0; // latest hello ping time
    private       static long deletionTime = 0;

    private static final int TIMELINE_PUSH_MINSIZE = 200;
    private static final int TIMELINE_PUSH_MAXSIZE = 1000;
    
    /**
     * ask the thread to shut down
     */
    public void shutdown() {
        this.shallRun = false;
        this.interrupt();
        DAO.log("catched caretaker termination signal");
    }
    
    @Override
    public void run() {
        Thread.currentThread().setName("CARETAKER");
        // send a message to other peers that I am alive
        String[] backends = DAO.getBackend();
        int maxRetries = Integer.valueOf(DAO.getConfig("caretaker.backendpush.retries", "5"));
        int backOffParameter = Integer.valueOf(DAO.getConfig("caretaker.backendpush.backoff", "3000"));

        boolean busy = false;
        // work loop
        beat: while (this.shallRun) try {
            // check upgrade time
            if (System.currentTimeMillis() > upgradeTime) {
                // increase the upgrade time to prevent that the peer runs amok (re-tries the attempt all the time) when upgrade fails for any reason
                upgradeTime = upgradeTime + upgradeWait;
                
                // do an upgrade
                DAO.log("UPGRADE: starting an upgrade");
                upgrade();
                DAO.log("UPGRADE: started an upgrade");
            }
            
            // check ping
            if (System.currentTimeMillis() - helloPeriod > helloTime) {
                helloTime = System.currentTimeMillis();
                HelloService.propagate(backends);
            }
            
            // clear caches
            if (SuggestServlet.cache.size() > 100) SuggestServlet.cache.clear();
            
            // sleep a bit to prevent that the DoS limit fires at backend server
            try {Thread.sleep(busy ? 500 : 10000);} catch (InterruptedException e) {}
            if (!this.shallRun) break beat;
            busy = false;
            
            //DAO.log("connection pool: " + ClientConnection.cm.getTotalStats().toString());
            
            // peer-to-peer operation
            TwitterTimeline tl = DAO.outgoingMessages.takeTimelineMin(Order.CREATED_AT, TIMELINE_PUSH_MINSIZE, TIMELINE_PUSH_MAXSIZE);
            int tl_size = tl == null ? 0 : tl.size();
            if (tl_size > 0 && backends.length > 0) {
                // transmit the timeline
                long start = System.currentTimeMillis();
                boolean success = PushServlet.push(backends, tl);
                if (success) {
                    DAO.log("success pushing " + tl.size() + " messages to backend " + Arrays.toString(backends) + " in 1st attempt in " + (System.currentTimeMillis() - start) + " ms");
                } else {
                    // we should try again.. but not an infinite number because then
                    // our timeline in RAM would fill up our RAM creating a memory leak
{'fraction_non_alphanumeric': 0.08956916099773243, 'fraction_numerical': 0.010487528344671202, 'mean_word_length': 2.433344145313007, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4797740', 'n_tokens_mistral': 3039, 'n_tokens_neox': 2891, 'n_words': 917}