1/*  Part of SWI-Prolog
    2
    3    Author:        Jan Wielemaker
    4    E-mail:        J.Wielemaker@vu.nl
    5    WWW:           http://www.swi-prolog.org
    6    Copyright (c)  2006-2015, University of Amsterdam
    7                              VU University Amsterdam
    8    All rights reserved.
    9
   10    Redistribution and use in source and binary forms, with or without
   11    modification, are permitted provided that the following conditions
   12    are met:
   13
   14    1. Redistributions of source code must retain the above copyright
   15       notice, this list of conditions and the following disclaimer.
   16
   17    2. Redistributions in binary form must reproduce the above copyright
   18       notice, this list of conditions and the following disclaimer in
   19       the documentation and/or other materials provided with the
   20       distribution.
   21
   22    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
   23    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
   24    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
   25    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
   26    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
   27    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
   28    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   29    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   30    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   31    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
   32    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   33    POSSIBILITY OF SUCH DAMAGE.
   34*/
   35
   36:- module(rdf_persistency,
   37          [ rdf_attach_db/2,            % +Directory, +Options
   38            rdf_detach_db/0,            % +Detach current Graph
   39            rdf_current_db/1,           % -Directory
   40            rdf_persistency/2,          % +Graph, +Bool
   41            rdf_flush_journals/1,       % +Options
   42            rdf_persistency_property/1, % ?Property
   43            rdf_journal_file/2,         % ?Graph, ?JournalFile
   44            rdf_snapshot_file/2,        % ?Graph, ?SnapshotFile
   45            rdf_db_to_file/2            % ?Graph, ?FileBase
   46          ]).   47:- use_module(library(semweb/rdf_db),
   48              [ rdf_graph/1, rdf_unload_graph/1, rdf_statistics/1,
   49                rdf_load_db/1, rdf_retractall/4, rdf_create_graph/1,
   50                rdf_assert/4, rdf_update/5, rdf_monitor/2, rdf/4,
   51                rdf_save_db/2, rdf_atom_md5/3, rdf_current_ns/2,
   52                rdf_register_ns/3
   53              ]).   54
   55:- autoload(library(apply),[maplist/2,maplist/3,partition/4,exclude/3]).   56:- use_module(library(debug),[debug/3]).   57:- autoload(library(error),
   58	    [permission_error/3,must_be/2,domain_error/2]).   59:- autoload(library(filesex),
   60	    [directory_file_path/3,make_directory_path/1]).   61:- autoload(library(lists),[select/3,append/3]).   62:- autoload(library(option),[option/2,option/3]).   63:- autoload(library(readutil),[read_file_to_terms/3]).   64:- autoload(library(socket),[gethostname/1]).   65:- autoload(library(thread),[concurrent/3]).   66:- autoload(library(uri),[uri_encoded/3]).

  100:- volatile
  101    rdf_directory/1,
  102    rdf_lock/2,
  103    rdf_option/1,
  104    source_journal_fd/2,
  105    file_base_db/2.  106:- dynamic
  107    rdf_directory/1,                % Absolute path
  108    rdf_lock/2,                     % Dir, Lock
  109    rdf_option/1,                   % Defined options
  110    source_journal_fd/2,            % DB, JournalFD
  111    file_base_db/2.                 % FileBase, DB
  112
  113:- meta_predicate
  114    no_agc(0).  115
  116:- predicate_options(rdf_attach_db/2, 2,
  117                     [ access(oneof([read_write,read_only])),
  118                       concurrency(positive_integer),
  119                       max_open_journals(positive_integer),
  120                       silent(oneof([true,false,brief])),
  121                       log_nested_transactions(boolean)
  122                     ]).

 rdf_attach_db(+Directory, +Options) is det

Start persistent operations using Directory as place to store files. There are several cases:

• Empty DB, existing directory Load the DB from the existing directory
• Full DB, empty directory Create snapshots for all sources in directory

Options:

access(+AccessMode)

One of auto (default), read_write or read_only. Read-only access implies that the RDF store is not locked. It is read at startup and all modifications to the data are temporary. The default auto mode is read_write if the directory is writeable and the lock can be acquired. Otherwise it reverts to read_only.

concurrency(+Jobs)

Number of threads to use for loading the initial database. If not provided it is the number of CPUs as obtained from the flag cpu_count.

max_open_journals(+Count)

Maximum number of journals kept open. If not provided, the default is 10. See limit_fd_pool/0.

directory_levels(+Count)

Number of levels of intermediate directories for storing the graph files. Default is 2.

silent(+BoolOrBrief)

If true (default false), do not print informational messages. Finally, if brief it will show minimal feedback.

log_nested_transactions(+Boolean)

If true, nested log transactions are added to the journal information. By default (false), no log-term is added for nested transactions.\\

Errors

- existence_error(source_sink, Directory)

- permission_error(write, directory, Directory)

  172rdf_attach_db(DirSpec, Options) :-
  173    option(access(read_only), Options),
  174    !,
  175    absolute_file_name(DirSpec,
  176                       Directory,
  177                       [ access(read),
  178                         file_type(directory)
  179                       ]),
  180    rdf_attach_db_ro(Directory, Options).
  181rdf_attach_db(DirSpec, Options) :-
  182    option(access(read_write), Options),
  183    !,
  184    rdf_attach_db_rw(DirSpec, Options).
  185rdf_attach_db(DirSpec, Options) :-
  186    absolute_file_name(DirSpec,
  187                       Directory,
  188                       [ access(exist),
  189                         file_type(directory),
  190                         file_errors(fail)
  191                       ]),
  192    !,
  193    (   access_file(Directory, write)
  194    ->  catch(rdf_attach_db_rw(Directory, Options), E, true),
  195        (   var(E)
  196        ->  true
  197        ;   E = error(permission_error(lock, rdf_db, _), _)
  198        ->  print_message(warning, E),
  199            print_message(warning, rdf(read_only)),
  200            rdf_attach_db(DirSpec, [access(read_only)|Options])
  201        ;   throw(E)
  202        )
  203    ;   print_message(warning,
  204                      error(permission_error(write, directory, Directory))),
  205        print_message(warning, rdf(read_only)),
  206        rdf_attach_db_ro(Directory, Options)
  207    ).
  208rdf_attach_db(DirSpec, Options) :-
  209    catch(rdf_attach_db_rw(DirSpec, Options), E, true),
  210    (   var(E)
  211    ->  true
  212    ;   print_message(warning, E),
  213        print_message(warning, rdf(read_only)),
  214        rdf_attach_db(DirSpec, [access(read_only)|Options])
  215    ).
  216
  217
  218rdf_attach_db_rw(DirSpec, Options) :-
  219    absolute_file_name(DirSpec,
  220                       Directory,
  221                       [ access(write),
  222                         file_type(directory),
  223                         file_errors(fail)
  224                       ]),
  225    !,
  226    (   rdf_directory(Directory)
  227    ->  true                        % update settings?
  228    ;   rdf_detach_db,
  229        mkdir(Directory),
  230        lock_db(Directory),
  231        assert(rdf_directory(Directory)),
  232        assert_options(Options),
  233        stop_monitor,               % make sure not to register load
  234        no_agc(load_db),
  235        at_halt(rdf_detach_db),
  236        start_monitor
  237    ).
  238rdf_attach_db_rw(DirSpec, Options) :-
  239    absolute_file_name(DirSpec,
  240                       Directory,
  241                       [ solutions(all)
  242                       ]),
  243    (   exists_directory(Directory)
  244    ->  access_file(Directory, write)
  245    ;   catch(make_directory(Directory), _, fail)
  246    ),
  247    !,
  248    rdf_attach_db(Directory, Options).
  249rdf_attach_db_rw(DirSpec, _) :-         % Generate an existence or
  250    absolute_file_name(DirSpec,     % permission error
  251                       Directory,
  252                       [ access(exist),
  253                         file_type(directory)
  254                       ]),
  255    permission_error(write, directory, Directory).

 rdf_attach_db_ro(+Directory, +Options)

Open an RDF database in read-only mode.

  261rdf_attach_db_ro(Directory, Options) :-
  262    rdf_detach_db,
  263    assert(rdf_directory(Directory)),
  264    assert_options(Options),
  265    stop_monitor,           % make sure not to register load
  266    no_agc(load_db).
  267
  268
  269assert_options([]).
  270assert_options([H|T]) :-
  271    (   option_type(H, Check)
  272    ->  Check,
  273        assert(rdf_option(H))
  274    ;   true                        % ignore options we do not understand
  275    ),
  276    assert_options(T).
  277
  278option_type(concurrency(X),             must_be(positive_integer, X)).
  279option_type(max_open_journals(X),       must_be(positive_integer, X)).
  280option_type(directory_levels(X),        must_be(positive_integer, X)).
  281option_type(silent(X),                  must_be(oneof([true,false,brief]), X)).
  282option_type(log_nested_transactions(X), must_be(boolean, X)).
  283option_type(access(X),                  must_be(oneof([read_write,
  284                                                       read_only]), X)).

 rdf_persistency_property(?Property) is nondet

True when Property is a property of the current persistent database. Exposes the properties that can be passed as options to rdf_attach_db/2. Specifically, rdf_persistency_property(access(read_only)) is true iff the database is mounted in read-only mode. In addition, the following property is supported:

directory(Dir)

The directory in which the database resides.

  299rdf_persistency_property(Property) :-
  300    var(Property),
  301    !,
  302    rdf_persistency_property_(Property).
  303rdf_persistency_property(Property) :-
  304    rdf_persistency_property_(Property),
  305    !.
  306
  307rdf_persistency_property_(Property) :-
  308    rdf_option(Property).
  309rdf_persistency_property_(directory(Dir)) :-
  310    rdf_directory(Dir).

 no_agc(:Goal)

Run Goal with atom garbage collection disabled. Loading an RDF database creates large amounts of atoms we know are not garbage.

  318no_agc(Goal) :-
  319    setup_call_cleanup(
  320        push_prolog_flag(agc_margin, 0),
  321        Goal,
  322        pop_prolog_flag(agc_margin)).

 rdf_detach_db is det

Detach from the current database. Succeeds silently if no database is attached. Normally called at the end of the program through at_halt/1.

  331rdf_detach_db :-
  332    debug(halt, 'Detaching RDF database', []),
  333    stop_monitor,
  334    close_journals,
  335    (   retract(rdf_directory(Dir))
  336    ->  debug(halt, 'DB Directory: ~w', [Dir]),
  337        save_prefixes(Dir),
  338        retractall(rdf_option(_)),
  339        retractall(source_journal_fd(_,_)),
  340        unlock_db(Dir)
  341    ;   true
  342    ).

 rdf_current_db(?Dir)

True if Dir is the current RDF persistent database.

  349rdf_current_db(Directory) :-
  350    rdf_directory(Dir),
  351    !,
  352    Dir = Directory.

 rdf_flush_journals(+Options)

Flush dirty journals. Options:

min_size(+KB)

Only flush if journal is over KB in size.

graph(+Graph)

Only flush the journal of Graph

To be done

- Provide a default for min_size?

  366rdf_flush_journals(Options) :-
  367    option(graph(Graph), Options, _),
  368    forall(rdf_graph(Graph),
  369           rdf_flush_journal(Graph, Options)).
  370
  371rdf_flush_journal(Graph, Options) :-
  372    db_files(Graph, _SnapshotFile, JournalFile),
  373    db_file(JournalFile, File),
  374    (   \+ exists_file(File)
  375    ->  true
  376    ;   memberchk(min_size(KB), Options),
  377        size_file(File, Size),
  378        Size / 1024 < KB
  379    ->  true
  380    ;   create_db(Graph)
  381    ).
  382
  383                 /*******************************
  384                 *             LOAD             *
  385                 *******************************/

 load_db is det

Reload database from the directory specified by rdf_directory/1. First we find all names graphs using find_dbs/1 and then we load them.

  393load_db :-
  394    rdf_directory(Dir),
  395    concurrency(Jobs),
  396    cpu_stat_key(Jobs, StatKey),
  397    get_time(Wall0),
  398    statistics(StatKey, T0),
  399    load_prefixes(Dir),
  400    verbosity(Silent),
  401    find_dbs(Dir, Graphs, SnapShots, Journals),
  402    length(Graphs, GraphCount),
  403    maplist(rdf_unload_graph, Graphs),
  404    rdf_statistics(triples(Triples0)),
  405    load_sources(snapshots, SnapShots, Silent, Jobs),
  406    load_sources(journals, Journals, Silent, Jobs),
  407    rdf_statistics(triples(Triples1)),
  408    statistics(StatKey, T1),
  409    get_time(Wall1),
  410    T is T1 - T0,
  411    Wall is Wall1 - Wall0,
  412    Triples = Triples1 - Triples0,
  413    message_level(Silent, Level),
  414    print_message(Level, rdf(restore(attached(GraphCount, Triples, T/Wall)))).
  415
  416load_sources(_, [], _, _) :- !.
  417load_sources(Type, Sources, Silent, Jobs) :-
  418    length(Sources, Count),
  419    RunJobs is min(Count, Jobs),
  420    print_message(informational, rdf(restoring(Type, Count, RunJobs))),
  421    make_goals(Sources, Silent, 1, Count, Goals),
  422    concurrent(RunJobs, Goals, []).

 make_goals(+DBs, +Silent, +Index, +Total, -Goals)

  427make_goals([], _, _, _, []).
  428make_goals([DB|T0], Silent, I,  Total,
  429           [load_source(DB, Silent, I, Total)|T]) :-
  430    I2 is I + 1,
  431    make_goals(T0, Silent, I2, Total, T).
  432
  433verbosity(Silent) :-
  434    rdf_option(silent(Silent)),
  435    !.
  436verbosity(Silent) :-
  437    current_prolog_flag(verbose, silent),
  438    !,
  439    Silent = true.
  440verbosity(brief).

 concurrency(-Jobs)

Number of jobs to run concurrently.

  447concurrency(Jobs) :-
  448    rdf_option(concurrency(Jobs)),
  449    !.
  450concurrency(Jobs) :-
  451    current_prolog_flag(cpu_count, Jobs),
  452    Jobs > 0,
  453    !.
  454concurrency(1).
  455
  456cpu_stat_key(1, cputime) :- !.
  457cpu_stat_key(_, process_cputime).

 find_dbs(+Dir, -Graphs, -SnapBySize, -JournalBySize) is det

Scan the persistent database and return a list of snapshots and journals, both sorted by file-size. Each term is of the form

db(Size, Ext, DB, DBFile, Depth)

  469find_dbs(Dir, Graphs, SnapBySize, JournalBySize) :-
  470    directory_files(Dir, Files),
  471    phrase(scan_db_files(Files, Dir, '.', 0), Scanned),
  472    maplist(db_graph, Scanned, UnsortedGraphs),
  473    sort(UnsortedGraphs, Graphs),
  474    (   consider_reindex_db(Dir, Graphs, Scanned)
  475    ->  find_dbs(Dir, Graphs, SnapBySize, JournalBySize)
  476    ;   partition(db_is_snapshot, Scanned, Snapshots, Journals),
  477        sort(Snapshots, SnapBySize),
  478        sort(Journals, JournalBySize)
  479    ).
  480
  481consider_reindex_db(Dir, Graphs, Scanned) :-
  482    length(Graphs, Count),
  483    Count > 0,
  484    DepthNeeded is floor(log(Count)/log(256)),
  485    (   maplist(depth_db(DepthNow), Scanned)
  486    ->  (   DepthNeeded > DepthNow
  487        ->  true
  488        ;   retractall(rdf_option(directory_levels(_))),
  489            assertz(rdf_option(directory_levels(DepthNow))),
  490            fail
  491        )
  492    ;   true
  493    ),
  494    reindex_db(Dir, DepthNeeded).
  495
  496db_is_snapshot(Term) :-
  497    arg(2, Term, trp).
  498
  499db_graph(Term, DB) :-
  500    arg(3, Term, DB).
  501
  502db_file_name(Term, File) :-
  503    arg(4, Term, File).
  504
  505depth_db(Depth, DB) :-
  506    arg(5, DB, Depth).

 scan_db_files(+Files, +Dir, +Prefix, +Depth)// is det

Produces a list of db(DB, Size, File) for all recognised RDF database files. File is relative to the database directory Dir.

  513scan_db_files([], _, _, _) -->
  514    [].
  515scan_db_files([Nofollow|T], Dir, Prefix, Depth) -->
  516    { nofollow(Nofollow) },
  517    !,
  518    scan_db_files(T, Dir, Prefix, Depth).
  519scan_db_files([File|T], Dir, Prefix, Depth) -->
  520    { file_name_extension(Base, Ext, File),
  521      db_extension(Ext),
  522      !,
  523      rdf_db_to_file(DB, Base),
  524      directory_file_path(Prefix, File, DBFile),
  525      directory_file_path(Dir, DBFile, AbsFile),
  526      size_file(AbsFile, Size)
  527    },
  528    [ db(Size, Ext, DB, AbsFile, Depth) ],
  529    scan_db_files(T, Dir, Prefix, Depth).
  530scan_db_files([D|T], Dir, Prefix, Depth) -->
  531    { directory_file_path(Prefix, D, SubD),
  532      directory_file_path(Dir, SubD, AbsD),
  533      exists_directory(AbsD),
  534      \+ read_link(AbsD, _, _),    % Do not follow links
  535      !,
  536      directory_files(AbsD, SubFiles),
  537      SubDepth is Depth + 1
  538    },
  539    scan_db_files(SubFiles, Dir, SubD, SubDepth),
  540    scan_db_files(T, Dir, Prefix, Depth).
  541scan_db_files([_|T], Dir, Prefix, Depth) -->
  542    scan_db_files(T, Dir, Prefix, Depth).
  543
  544nofollow(.).
  545nofollow(..).
  546
  547db_extension(trp).
  548db_extension(jrn).
  549
  550:- public load_source/4.                % called through make_goals/5
  551
  552load_source(DB, Silent, Nth, Total) :-
  553    db_file_name(DB, File),
  554    db_graph(DB, Graph),
  555    message_level(Silent, Level),
  556    graph_triple_count(Graph, Count0),
  557    statistics(cputime, T0),
  558    (   db_is_snapshot(DB)
  559    ->  print_message(Level, rdf(restore(Silent, snapshot(Graph, File)))),
  560        rdf_load_db(File)
  561    ;   print_message(Level, rdf(restore(Silent, journal(Graph, File)))),
  562        load_journal(File, Graph)
  563    ),
  564    statistics(cputime, T1),
  565    T is T1 - T0,
  566    graph_triple_count(Graph, Count1),
  567    Count is Count1 - Count0,
  568    print_message(Level, rdf(restore(Silent,
  569                                     done(Graph, T, Count, Nth, Total)))).
  570
  571
  572graph_triple_count(Graph, Count) :-
  573    rdf_statistics(triples_by_graph(Graph, Count)),
  574    !.
  575graph_triple_count(_, 0).

 attach_graph(+Graph, +Options) is det

Load triples and reload journal from the indicated snapshot file.

  583attach_graph(Graph, Options) :-
  584    (   option(silent(true), Options)
  585    ->  Level = silent
  586    ;   Level = informational
  587    ),
  588    db_files(Graph, SnapshotFile, JournalFile),
  589    rdf_retractall(_,_,_,Graph),
  590    statistics(cputime, T0),
  591    print_message(Level, rdf(restore(Silent, Graph))),
  592    db_file(SnapshotFile, AbsSnapShot),
  593    (   exists_file(AbsSnapShot)
  594    ->  print_message(Level, rdf(restore(Silent, snapshot(SnapshotFile)))),
  595        rdf_load_db(AbsSnapShot)
  596    ;   true
  597    ),
  598    (   exists_db(JournalFile)
  599    ->  print_message(Level, rdf(restore(Silent, journal(JournalFile)))),
  600        load_journal(JournalFile, Graph)
  601    ;   true
  602    ),
  603    statistics(cputime, T1),
  604    T is T1 - T0,
  605    (   rdf_statistics(triples_by_graph(Graph, Count))
  606    ->  true
  607    ;   Count = 0
  608    ),
  609    print_message(Level, rdf(restore(Silent,
  610                                     done(Graph, T, Count)))).
  611
  612message_level(true, silent) :- !.
  613message_level(_, informational).
  614
  615
  616                 /*******************************
  617                 *         LOAD JOURNAL         *
  618                 *******************************/

 load_journal(+File:atom, +DB:atom) is det

Process transactions from the RDF journal File, adding the given named graph.

  625load_journal(File, DB) :-
  626    rdf_create_graph(DB),
  627    setup_call_cleanup(
  628        open(File, read, In, [encoding(utf8)]),
  629        ( read(In, T0),
  630          process_journal(T0, In, DB)
  631        ),
  632        close(In)).
  633
  634process_journal(end_of_file, _, _) :- !.
  635process_journal(Term, In, DB) :-
  636    (   process_journal_term(Term, DB)
  637    ->  true
  638    ;   throw(error(type_error(journal_term, Term), _))
  639    ),
  640    read(In, T2),
  641    process_journal(T2, In, DB).
  642
  643process_journal_term(assert(S,P,O), DB) :-
  644    rdf_assert(S,P,O,DB).
  645process_journal_term(assert(S,P,O,Line), DB) :-
  646    rdf_assert(S,P,O,DB:Line).
  647process_journal_term(retract(S,P,O), DB) :-
  648    rdf_retractall(S,P,O,DB).
  649process_journal_term(retract(S,P,O,Line), DB) :-
  650    rdf_retractall(S,P,O,DB:Line).
  651process_journal_term(update(S,P,O,Action), DB) :-
  652    (   rdf_update(S,P,O,DB, Action)
  653    ->  true
  654    ;   print_message(warning, rdf(update_failed(S,P,O,Action)))
  655    ).
  656process_journal_term(start(_), _).      % journal open/close
  657process_journal_term(end(_), _).
  658process_journal_term(begin(_), _).      % logged transaction (compatibility)
  659process_journal_term(end, _).
  660process_journal_term(begin(_,_,_,_), _). % logged transaction (current)
  661process_journal_term(end(_,_,_), _).
  662
  663
  664                 /*******************************
  665                 *         CREATE JOURNAL       *
  666                 *******************************/
  667
  668:- dynamic
  669    blocked_db/2,                   % DB, Reason
  670    transaction_message/3,          % Nesting, Time, Message
  671    transaction_db/3.               % Nesting, DB, Id

 rdf_persistency(+DB, Bool)

Specify whether a database is persistent. Switching to false kills the persistent state. Switching to true creates it.

  678rdf_persistency(DB, Bool) :-
  679    must_be(atom, DB),
  680    must_be(boolean, Bool),
  681    fail.
  682rdf_persistency(DB, false) :-
  683    !,
  684    (   blocked_db(DB, persistency)
  685    ->  true
  686    ;   assert(blocked_db(DB, persistency)),
  687        delete_db(DB)
  688    ).
  689rdf_persistency(DB, true) :-
  690    (   retract(blocked_db(DB, persistency))
  691    ->  create_db(DB)
  692    ;   true
  693    ).

 rdf_db:property_of_graph(?Property, +Graph) is nondet

Extend rdf_graph_property/2 with new properties.

  699:- multifile
  700    rdf_db:property_of_graph/2.  701
  702rdf_db:property_of_graph(persistent(State), Graph) :-
  703    (   blocked_db(Graph, persistency)
  704    ->  State = false
  705    ;   State = true
  706    ).

 start_monitor is det

 stop_monitor is det

Start/stop monitoring the RDF database for changes and update the journal.

  715start_monitor :-
  716    rdf_monitor(monitor,
  717                [ -assert(load)
  718                ]).
  719stop_monitor :-
  720    rdf_monitor(monitor,
  721                [ -all
  722                ]).

 monitor(+Term) is semidet

Handle an rdf_monitor/2 callback to deal with persistency. Note that the monitor calls that come from rdf_db.pl that deal with database changes are serialized. They do come from different threads though.

  731monitor(Msg) :-
  732    debug(monitor, 'Monitor: ~p~n', [Msg]),
  733    fail.
  734monitor(assert(S,P,O,DB:Line)) :-
  735    !,
  736    \+ blocked_db(DB, _),
  737    journal_fd(DB, Fd),
  738    open_transaction(DB, Fd),
  739    format(Fd, '~q.~n', [assert(S,P,O,Line)]),
  740    sync_journal(DB, Fd).
  741monitor(assert(S,P,O,DB)) :-
  742    \+ blocked_db(DB, _),
  743    journal_fd(DB, Fd),
  744    open_transaction(DB, Fd),
  745    format(Fd, '~q.~n', [assert(S,P,O)]),
  746    sync_journal(DB, Fd).
  747monitor(retract(S,P,O,DB:Line)) :-
  748    !,
  749    \+ blocked_db(DB, _),
  750    journal_fd(DB, Fd),
  751    open_transaction(DB, Fd),
  752    format(Fd, '~q.~n', [retract(S,P,O,Line)]),
  753    sync_journal(DB, Fd).
  754monitor(retract(S,P,O,DB)) :-
  755    \+ blocked_db(DB, _),
  756    journal_fd(DB, Fd),
  757    open_transaction(DB, Fd),
  758    format(Fd, '~q.~n', [retract(S,P,O)]),
  759    sync_journal(DB, Fd).
  760monitor(update(S,P,O,DB:Line,Action)) :-
  761    !,
  762    \+ blocked_db(DB, _),
  763    (   Action = graph(NewDB)
  764    ->  monitor(assert(S,P,O,NewDB)),
  765        monitor(retract(S,P,O,DB:Line))
  766    ;   journal_fd(DB, Fd),
  767        format(Fd, '~q.~n', [update(S,P,O,Action)]),
  768        sync_journal(DB, Fd)
  769    ).
  770monitor(update(S,P,O,DB,Action)) :-
  771    \+ blocked_db(DB, _),
  772    (   Action = graph(NewDB)
  773    ->  monitor(assert(S,P,O,NewDB)),
  774        monitor(retract(S,P,O,DB))
  775    ;   journal_fd(DB, Fd),
  776        open_transaction(DB, Fd),
  777        format(Fd, '~q.~n', [update(S,P,O,Action)]),
  778        sync_journal(DB, Fd)
  779    ).
  780monitor(load(BE, _DumpFileURI)) :-
  781    (   BE = end(Graphs)
  782    ->  sync_loaded_graphs(Graphs)
  783    ;   true
  784    ).
  785monitor(create_graph(Graph)) :-
  786    \+ blocked_db(Graph, _),
  787    journal_fd(Graph, Fd),
  788    open_transaction(Graph, Fd),
  789    sync_journal(Graph, Fd).
  790monitor(reset) :-
  791    forall(rdf_graph(Graph), delete_db(Graph)).
  792                                        % TBD: Remove empty directories?
  793
  794monitor(transaction(BE, Id)) :-
  795    monitor_transaction(Id, BE).
  796
  797monitor_transaction(load_journal(DB), begin(_)) :-
  798    !,
  799    assert(blocked_db(DB, journal)).
  800monitor_transaction(load_journal(DB), end(_)) :-
  801    !,
  802    retractall(blocked_db(DB, journal)).
  803
  804monitor_transaction(parse(URI), begin(_)) :-
  805    !,
  806    (   blocked_db(URI, persistency)
  807    ->  true
  808    ;   assert(blocked_db(URI, parse))
  809    ).
  810monitor_transaction(parse(URI), end(_)) :-
  811    !,
  812    (   retract(blocked_db(URI, parse))
  813    ->  create_db(URI)
  814    ;   true
  815    ).
  816monitor_transaction(unload(DB), begin(_)) :-
  817    !,
  818    (   blocked_db(DB, persistency)
  819    ->  true
  820    ;   assert(blocked_db(DB, unload))
  821    ).
  822monitor_transaction(unload(DB), end(_)) :-
  823    !,
  824    (   retract(blocked_db(DB, unload))
  825    ->  delete_db(DB)
  826    ;   true
  827    ).
  828monitor_transaction(log(Msg), begin(N)) :-
  829    !,
  830    check_nested(N),
  831    get_time(Time),
  832    asserta(transaction_message(N, Time, Msg)).
  833monitor_transaction(log(_), end(N)) :-
  834    check_nested(N),
  835    retract(transaction_message(N, _, _)),
  836    !,
  837    findall(DB:Id, retract(transaction_db(N, DB, Id)), DBs),
  838    end_transactions(DBs, N).
  839monitor_transaction(log(Msg, DB), begin(N)) :-
  840    !,
  841    check_nested(N),
  842    get_time(Time),
  843    asserta(transaction_message(N, Time, Msg)),
  844    journal_fd(DB, Fd),
  845    open_transaction(DB, Fd).
  846monitor_transaction(log(Msg, _DB), end(N)) :-
  847    monitor_transaction(log(Msg), end(N)).

 check_nested(+Level) is semidet

True if we must log this transaction. This is always the case for toplevel transactions. Nested transactions are only logged if log_nested_transactions(true) is defined.

  856check_nested(0) :- !.
  857check_nested(_) :-
  858    rdf_option(log_nested_transactions(true)).

 open_transaction(+DB, +Fd) is det

Add a begin(Id, Level, Time, Message) term if a transaction involves DB. Id is an incremental integer, where each database has its own counter. Level is the nesting level, Time is a floating point timestamp and Message is the message provided as argument to the log message.

  869open_transaction(DB, Fd) :-
  870    transaction_message(N, Time, Msg),
  871    !,
  872    (   transaction_db(N, DB, _)
  873    ->  true
  874    ;   next_transaction_id(DB, Id),
  875        assert(transaction_db(N, DB, Id)),
  876        RoundedTime is round(Time*100)/100,
  877        format(Fd, '~q.~n', [begin(Id, N, RoundedTime, Msg)])
  878    ).
  879open_transaction(_,_).

 next_transaction_id(+DB, -Id) is det

Id is the number to user for the next logged transaction on DB. Transactions in each named graph are numbered in sequence. Searching the Id of the last transaction is performed by the 2nd clause starting 1Kb from the end and doubling this offset each failure.

  890:- dynamic
  891    current_transaction_id/2.  892
  893next_transaction_id(DB, Id) :-
  894    retract(current_transaction_id(DB, Last)),
  895    !,
  896    Id is Last + 1,
  897    assert(current_transaction_id(DB, Id)).
  898next_transaction_id(DB, Id) :-
  899    db_files(DB, _, Journal),
  900    exists_file(Journal),
  901    !,
  902    size_file(Journal, Size),
  903    open_db(Journal, read, In, []),
  904    call_cleanup(iterative_expand(In, Size, Last), close(In)),
  905    Id is Last + 1,
  906    assert(current_transaction_id(DB, Id)).
  907next_transaction_id(DB, 1) :-
  908    assert(current_transaction_id(DB, 1)).
  909
  910iterative_expand(_, 0, 0) :- !.
  911iterative_expand(In, Size, Last) :-     % Scan growing sections from the end
  912    Max is floor(log(Size)/log(2)),
  913    between(10, Max, Step),
  914    Offset is -(1<<Step),
  915    seek(In, Offset, eof, _),
  916    skip(In, 10),                   % records are line-based
  917    read(In, T0),
  918    last_transaction_id(T0, In, 0, Last),
  919    Last > 0,
  920    !.
  921iterative_expand(In, _, Last) :-        % Scan the whole file
  922    seek(In, 0, bof, _),
  923    read(In, T0),
  924    last_transaction_id(T0, In, 0, Last).
  925
  926last_transaction_id(end_of_file, _, Last, Last) :- !.
  927last_transaction_id(end(Id, _, _), In, _, Last) :-
  928    read(In, T1),
  929    last_transaction_id(T1, In, Id, Last).
  930last_transaction_id(_, In, Id, Last) :-
  931    read(In, T1),
  932    last_transaction_id(T1, In, Id, Last).

 end_transactions(+DBs:list(atom:id)) is det

End a transaction that affected the given list of databases. We write the list of other affected databases as an argument to the end-term to facilitate fast finding of the related transactions.

In each database, the transaction is ended with a term end(Id, Nesting, Others), where Id and Nesting are the transaction identifier and nesting (see open_transaction/2) and Others is a list of DB:Id, indicating other databases affected by the transaction.

  947end_transactions(DBs, N) :-
  948    end_transactions(DBs, DBs, N).
  949
  950end_transactions([], _, _).
  951end_transactions([DB:Id|T], DBs, N) :-
  952    journal_fd(DB, Fd),
  953    once(select(DB:Id, DBs, Others)),
  954    format(Fd, 'end(~q, ~q, ~q).~n', [Id, N, Others]),
  955    sync_journal(DB, Fd),
  956    end_transactions(T, DBs, N).

 sync_loaded_graphs(+Graphs)

Called after a binary triple has been loaded that added triples to the given graphs.

  964sync_loaded_graphs(Graphs) :-
  965    maplist(create_db, Graphs).
  966
  967
  968                 /*******************************
  969                 *         JOURNAL FILES        *
  970                 *******************************/

 journal_fd(+DB, -Stream) is det

Get an open stream to a journal. If the journal is not open, old journals are closed to satisfy the max_open_journals option. Then the journal is opened in append mode. Journal files are always encoded as UTF-8 for portability as well as to ensure full coverage of Unicode.

  980journal_fd(DB, Fd) :-
  981    source_journal_fd(DB, Fd),
  982    !.
  983journal_fd(DB, Fd) :-
  984    with_mutex(rdf_journal_file,
  985               journal_fd_(DB, Out)),
  986    Fd = Out.
  987
  988journal_fd_(DB, Fd) :-
  989    source_journal_fd(DB, Fd),
  990    !.
  991journal_fd_(DB, Fd) :-
  992    limit_fd_pool,
  993    db_files(DB, _Snapshot, Journal),
  994    open_db(Journal, append, Fd,
  995            [ close_on_abort(false)
  996            ]),
  997    time_stamp(Now),
  998    format(Fd, '~q.~n', [start([time(Now)])]),
  999    assert(source_journal_fd(DB, Fd)).              % new one at the end

 limit_fd_pool is det

Limit the number of open journals to max_open_journals (10). Note that calls from rdf_monitor/2 are issued in different threads, but as they are part of write operations they are fully synchronised.

 1008limit_fd_pool :-
 1009    predicate_property(source_journal_fd(_, _), number_of_clauses(N)),
 1010    !,
 1011    (   rdf_option(max_open_journals(Max))
 1012    ->  true
 1013    ;   Max = 10
 1014    ),
 1015    Close is N - Max,
 1016    forall(between(1, Close, _),
 1017           close_oldest_journal).
 1018limit_fd_pool.
 1019
 1020close_oldest_journal :-
 1021    source_journal_fd(DB, _Fd),
 1022    !,
 1023    debug(rdf_persistency, 'Closing old journal for ~q', [DB]),
 1024    close_journal(DB).
 1025close_oldest_journal.

 sync_journal(+DB, +Fd)

Sync journal represented by database and stream. If the DB is involved in a transaction there is no point flushing until the end of the transaction.

 1034sync_journal(DB, _) :-
 1035    transaction_db(_, DB, _),
 1036    !.
 1037sync_journal(_, Fd) :-
 1038    flush_output(Fd).

 close_journal(+DB) is det

Close the journal associated with DB if it is open.

 1044close_journal(DB) :-
 1045    with_mutex(rdf_journal_file,
 1046               close_journal_(DB)).
 1047
 1048close_journal_(DB) :-
 1049    (   retract(source_journal_fd(DB, Fd))
 1050    ->  time_stamp(Now),
 1051        format(Fd, '~q.~n', [end([time(Now)])]),
 1052        close(Fd, [force(true)])
 1053    ;   true
 1054    ).

 close_journals

Close all open journals.

 1060close_journals :-
 1061    forall(source_journal_fd(DB, _),
 1062           catch(close_journal(DB), E,
 1063                 print_message(error, E))).

 create_db(+Graph)

Create a saved version of Graph in corresponding file, close and delete journals.

 1070create_db(Graph) :-
 1071    \+ rdf(_,_,_,Graph),
 1072    !,
 1073    debug(rdf_persistency, 'Deleting empty Graph ~w', [Graph]),
 1074    delete_db(Graph).
 1075create_db(Graph) :-
 1076    debug(rdf_persistency, 'Saving Graph ~w', [Graph]),
 1077    close_journal(Graph),
 1078    db_abs_files(Graph, Snapshot, Journal),
 1079    atom_concat(Snapshot, '.new', NewSnapshot),
 1080    (   catch(( create_directory_levels(Snapshot),
 1081                rdf_save_db(NewSnapshot, Graph)
 1082              ), Error,
 1083              ( print_message(warning, Error),
 1084                fail
 1085              ))
 1086    ->  (   exists_file(Journal)
 1087        ->  delete_file(Journal)
 1088        ;   true
 1089        ),
 1090        rename_file(NewSnapshot, Snapshot),
 1091        debug(rdf_persistency, 'Saved Graph ~w', [Graph])
 1092    ;   catch(delete_file(NewSnapshot), _, true)
 1093    ).

 delete_db(+DB)

Remove snapshot and journal file for DB.

 1100delete_db(DB) :-
 1101    with_mutex(rdf_journal_file,
 1102               delete_db_(DB)).
 1103
 1104delete_db_(DB) :-
 1105    close_journal_(DB),
 1106    db_abs_files(DB, Snapshot, Journal),
 1107    !,
 1108    (   exists_file(Journal)
 1109    ->  delete_file(Journal)
 1110    ;   true
 1111    ),
 1112    (   exists_file(Snapshot)
 1113    ->  delete_file(Snapshot)
 1114    ;   true
 1115    ).
 1116delete_db_(_).
 1117
 1118                 /*******************************
 1119                 *             LOCKING          *
 1120                 *******************************/

 lock_db(+Dir)

Lock the database directory Dir.

 1126lock_db(Dir) :-
 1127    lockfile(Dir, File),
 1128    catch(open(File, update, Out, [lock(write), wait(false)]),
 1129          error(permission_error(Access, _, _), _),
 1130          locked_error(Access, Dir)),
 1131    (   current_prolog_flag(pid, PID)
 1132    ->  true
 1133    ;   PID = 0                     % TBD: Fix in Prolog
 1134    ),
 1135    time_stamp(Now),
 1136    gethostname(Host),
 1137    format(Out, '/* RDF Database is in use */~n~n', []),
 1138    format(Out, '~q.~n', [ locked([ time(Now),
 1139                                    pid(PID),
 1140                                    host(Host)
 1141                                  ])
 1142                         ]),
 1143    flush_output(Out),
 1144    set_end_of_stream(Out),
 1145    assert(rdf_lock(Dir, lock(Out, File))),
 1146    at_halt(unlock_db(Dir)).
 1147
 1148locked_error(lock, Dir) :-
 1149    lockfile(Dir, File),
 1150    (   catch(read_file_to_terms(File, Terms, []), _, fail),
 1151        Terms = [locked(Args)]
 1152    ->  Context = rdf_locked(Args)
 1153    ;   Context = context(_, 'Database is in use')
 1154    ),
 1155    throw(error(permission_error(lock, rdf_db, Dir), Context)).
 1156locked_error(open, Dir) :-
 1157    throw(error(permission_error(lock, rdf_db, Dir),
 1158                context(_, 'Lock file cannot be opened'))).

 unlock_db(+Dir) is det

 unlock_db(+Stream, +File) is det

 1163unlock_db(Dir) :-
 1164    retract(rdf_lock(Dir, lock(Out, File))),
 1165    !,
 1166    unlock_db(Out, File).
 1167unlock_db(_).
 1168
 1169unlock_db(Out, File) :-
 1170    close(Out),
 1171    delete_file(File).
 1172
 1173                 /*******************************
 1174                 *           FILENAMES          *
 1175                 *******************************/
 1176
 1177lockfile(Dir, LockFile) :-
 1178    atomic_list_concat([Dir, /, lock], LockFile).
 1179
 1180directory_levels(Levels) :-
 1181    rdf_option(directory_levels(Levels)),
 1182    !.
 1183directory_levels(2).
 1184
 1185db_file(Base, File) :-
 1186    rdf_directory(Dir),
 1187    directory_levels(Levels),
 1188    db_file(Dir, Base, Levels, File).
 1189
 1190db_file(Dir, Base, Levels, File) :-
 1191    dir_levels(Base, Levels, Segments, [Base]),
 1192    atomic_list_concat([Dir|Segments], /, File).
 1193
 1194open_db(Base, Mode, Stream, Options) :-
 1195    db_file(Base, File),
 1196    create_directory_levels(File),
 1197    open(File, Mode, Stream, [encoding(utf8)|Options]).
 1198
 1199create_directory_levels(_File) :-
 1200    rdf_option(directory_levels(0)),
 1201    !.
 1202create_directory_levels(File) :-
 1203    file_directory_name(File, Dir),
 1204    make_directory_path(Dir).
 1205
 1206exists_db(Base) :-
 1207    db_file(Base, File),
 1208    exists_file(File).

 dir_levels(+File, +Levels, ?Segments, ?Tail) is det

Create a list of intermediate directory names for File. Each directory consists of two hexadecimal digits.

 1215dir_levels(_, 0, Segments, Segments) :- !.
 1216dir_levels(File, Levels, Segments, Tail) :-
 1217    rdf_atom_md5(File, 1, Hash),
 1218    create_dir_levels(Levels, 0, Hash, Segments, Tail).
 1219
 1220create_dir_levels(0, _, _, Segments, Segments) :- !.
 1221create_dir_levels(N, S, Hash, [S1|Segments0], Tail) :-
 1222    sub_atom(Hash, S, 2, _, S1),
 1223    S2 is S+2,
 1224    N2 is N-1,
 1225    create_dir_levels(N2, S2, Hash, Segments0, Tail).

 db_files(+DB, -Snapshot, -Journal)

db_files(-DB, +Snapshot, -Journal)

db_files(-DB, -Snapshot, +Journal)

True if named graph DB is represented by the files Snapshot and Journal. The filenames are local to the directory representing the store.

 1236db_files(DB, Snapshot, Journal) :-
 1237    nonvar(DB),
 1238    !,
 1239    rdf_db_to_file(DB, Base),
 1240    atom_concat(Base, '.trp', Snapshot),
 1241    atom_concat(Base, '.jrn', Journal).
 1242db_files(DB, Snapshot, Journal) :-
 1243    nonvar(Snapshot),
 1244    !,
 1245    atom_concat(Base, '.trp', Snapshot),
 1246    atom_concat(Base, '.jrn', Journal),
 1247    rdf_db_to_file(DB, Base).
 1248db_files(DB, Snapshot, Journal) :-
 1249    nonvar(Journal),
 1250    !,
 1251    atom_concat(Base, '.jrn', Journal),
 1252    atom_concat(Base, '.trp', Snapshot),
 1253    rdf_db_to_file(DB, Base).
 1254
 1255db_abs_files(DB, Snapshot, Journal) :-
 1256    db_files(DB, Snapshot0, Journal0),
 1257    db_file(Snapshot0, Snapshot),
 1258    db_file(Journal0, Journal).

 rdf_journal_file(+Graph, -File) is semidet

rdf_journal_file(-Graph, -File) is nondet

True if File the name of the existing journal file for Graph.

 1266rdf_journal_file(Graph, Journal) :-
 1267    (   var(Graph)
 1268    ->  rdf_graph(Graph)
 1269    ;   true
 1270    ),
 1271    db_abs_files(Graph, _Snapshot, Journal),
 1272    exists_file(Journal).

 rdf_snapshot_file(+Graph, -File) is semidet

rdf_snapshot_file(-Graph, -File) is nondet

True if File the name of the existing snapshot file for Graph.

 1280rdf_snapshot_file(Graph, Snapshot) :-
 1281    (   var(Graph)
 1282    ->  rdf_graph(Graph)    % also pick the empty graphs
 1283    ;   true
 1284    ),
 1285    db_abs_files(Graph, Snapshot, _Journal),
 1286    exists_file(Snapshot).

 rdf_db_to_file(+DB, -File) is det

rdf_db_to_file(-DB, +File) is det

Translate between database encoding (often an file or URL) and the name we store in the directory. We keep a cache for two reasons. Speed, but much more important is that the mapping of raw --> encoded provided by www_form_encode/2 is not guaranteed to be unique by the W3C standards.

 1298rdf_db_to_file(DB, File) :-
 1299    file_base_db(File, DB),
 1300    !.
 1301rdf_db_to_file(DB, File) :-
 1302    url_to_filename(DB, File),
 1303    assert(file_base_db(File, DB)).

 url_to_filename(+URL, -FileName) is det

url_to_filename(-URL, +FileName) is det

Turn a valid URL into a filename. Earlier versions used www_form_encode/2, but this can produce characters that are not valid in filenames. We will use the same encoding as www_form_encode/2, but using our own rules for allowed characters. The only requirement is that we avoid any filename special character in use. The current encoding use US-ASCII alnum characters, _ and %

 1316url_to_filename(URL, FileName) :-
 1317    atomic(URL),
 1318    !,
 1319    atom_codes(URL, Codes),
 1320    phrase(url_encode(EncCodes), Codes),
 1321    atom_codes(FileName, EncCodes).
 1322url_to_filename(URL, FileName) :-
 1323    uri_encoded(path, URL, FileName).
 1324
 1325url_encode([0'+|T]) -->
 1326    " ",
 1327    !,
 1328    url_encode(T).
 1329url_encode([C|T]) -->
 1330    alphanum(C),
 1331    !,
 1332    url_encode(T).
 1333url_encode([C|T]) -->
 1334    no_enc_extra(C),
 1335    !,
 1336    url_encode(T).
 1337url_encode(Enc) -->
 1338    (   "\r\n"
 1339    ;   "\n"
 1340    ),
 1341    !,
 1342    { string_codes("%0D%0A", Codes),
 1343      append(Codes, T, Enc)
 1344    },
 1345    url_encode(T).
 1346url_encode([]) -->
 1347    eos,
 1348    !.
 1349url_encode([0'%,D1,D2|T]) -->
 1350    [C],
 1351    { Dv1 is (C>>4 /\ 0xf),
 1352      Dv2 is (C /\ 0xf),
 1353      code_type(D1, xdigit(Dv1)),
 1354      code_type(D2, xdigit(Dv2))
 1355    },
 1356    url_encode(T).
 1357
 1358eos([], []).
 1359
 1360alphanum(C) -->
 1361    [C],
 1362    { C < 128,                      % US-ASCII
 1363      code_type(C, alnum)
 1364    }.
 1365
 1366no_enc_extra(0'_) --> "_".
 1367
 1368
 1369                 /*******************************
 1370                 *             REINDEX          *
 1371                 *******************************/

 reindex_db(+Dir, +Levels)

Reindex the database by creating intermediate directories.

 1377reindex_db(Dir, Levels) :-
 1378    directory_files(Dir, Files),
 1379    reindex_files(Files, Dir, '.', 0, Levels),
 1380    remove_empty_directories(Files, Dir).
 1381
 1382reindex_files([], _, _, _, _).
 1383reindex_files([Nofollow|Files], Dir, Prefix, CLevel, Levels) :-
 1384    nofollow(Nofollow),
 1385    !,
 1386    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1387reindex_files([File|Files], Dir, Prefix, CLevel, Levels) :-
 1388    CLevel \== Levels,
 1389    file_name_extension(_Base, Ext, File),
 1390    db_extension(Ext),
 1391    !,
 1392    directory_file_path(Prefix, File, DBFile),
 1393    directory_file_path(Dir, DBFile, OldPath),
 1394    db_file(Dir, File, Levels, NewPath),
 1395    debug(rdf_persistency, 'Rename ~q --> ~q', [OldPath, NewPath]),
 1396    file_directory_name(NewPath, NewDir),
 1397    make_directory_path(NewDir),
 1398    rename_file(OldPath, NewPath),
 1399    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1400reindex_files([D|Files], Dir, Prefix, CLevel, Levels) :-
 1401    directory_file_path(Prefix, D, SubD),
 1402    directory_file_path(Dir, SubD, AbsD),
 1403    exists_directory(AbsD),
 1404    \+ read_link(AbsD, _, _),      % Do not follow links
 1405    !,
 1406    directory_files(AbsD, SubFiles),
 1407    CLevel2 is CLevel + 1,
 1408    reindex_files(SubFiles, Dir, SubD, CLevel2, Levels),
 1409    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1410reindex_files([_|Files], Dir, Prefix, CLevel, Levels) :-
 1411    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1412
 1413
 1414remove_empty_directories([], _).
 1415remove_empty_directories([File|Files], Dir) :-
 1416    \+ nofollow(File),
 1417    directory_file_path(Dir, File, Path),
 1418    exists_directory(Path),
 1419    \+ read_link(Path, _, _),
 1420    !,
 1421    directory_files(Path, Content),
 1422    exclude(nofollow, Content, RealContent),
 1423    (   RealContent == []
 1424    ->  debug(rdf_persistency, 'Remove empty dir ~q', [Path]),
 1425        delete_directory(Path)
 1426    ;   remove_empty_directories(RealContent, Path)
 1427    ),
 1428    remove_empty_directories(Files, Dir).
 1429remove_empty_directories([_|Files], Dir) :-
 1430    remove_empty_directories(Files, Dir).
 1431
 1432
 1433                 /*******************************
 1434                 *            PREFIXES          *
 1435                 *******************************/
 1436
 1437save_prefixes(Dir) :-
 1438    atomic_list_concat([Dir, /, 'prefixes.db'], PrefixFile),
 1439    setup_call_cleanup(open(PrefixFile, write, Out, [encoding(utf8)]),
 1440                       write_prefixes(Out),
 1441                       close(Out)).
 1442
 1443write_prefixes(Out) :-
 1444    format(Out, '% Snapshot of defined RDF prefixes~n~n', []),
 1445    forall(rdf_current_ns(Alias, URI),
 1446           format(Out, 'prefix(~q, ~q).~n', [Alias, URI])).

 load_prefixes(+RDFDBDir) is det

If the file RDFDBDir/prefixes.db exists, load the prefixes. The prefixes are registered using rdf_register_ns/3. Possible errors because the prefix definitions have changed are printed as warnings, retaining the old definition. Note that changing prefixes generally requires reloading all RDF from the source.

 1456load_prefixes(Dir) :-
 1457    atomic_list_concat([Dir, /, 'prefixes.db'], PrefixFile),
 1458    (   exists_file(PrefixFile)
 1459    ->  setup_call_cleanup(open(PrefixFile, read, In, [encoding(utf8)]),
 1460                           read_prefixes(In),
 1461                           close(In))
 1462    ;   true
 1463    ).
 1464
 1465read_prefixes(Stream) :-
 1466    read_term(Stream, T0, []),
 1467    read_prefixes(T0, Stream).
 1468
 1469read_prefixes(end_of_file, _) :- !.
 1470read_prefixes(prefix(Alias, URI), Stream) :-
 1471    !,
 1472    must_be(atom, Alias),
 1473    must_be(atom, URI),
 1474    catch(rdf_register_ns(Alias, URI, []), E,
 1475          print_message(warning, E)),
 1476    read_term(Stream, T, []),
 1477    read_prefixes(T, Stream).
 1478read_prefixes(Term, _) :-
 1479    domain_error(prefix_term, Term).
 1480
 1481
 1482                 /*******************************
 1483                 *              UTIL            *
 1484                 *******************************/

 mkdir(+Directory)

Create a directory if it does not already exist.

 1490mkdir(Directory) :-
 1491    exists_directory(Directory),
 1492    !.
 1493mkdir(Directory) :-
 1494    make_directory(Directory).

 time_stamp(-Integer)

Return time-stamp rounded to integer.

 1500time_stamp(Int) :-
 1501    get_time(Now),
 1502    Int is round(Now).
 1503
 1504
 1505                 /*******************************
 1506                 *            MESSAGES          *
 1507                 *******************************/
 1508
 1509:- multifile
 1510    prolog:message/3,
 1511    prolog:message_context/3. 1512
 1513prolog:message(rdf(Term)) -->
 1514    message(Term).
 1515
 1516message(restoring(Type, Count, Jobs)) -->
 1517    [ 'Restoring ~D ~w using ~D concurrent workers'-[Count, Type, Jobs] ].
 1518message(restore(attached(Graphs, Triples, Time/Wall))) -->
 1519    { catch(Percent is round(100*Time/Wall), _, Percent = 0) },
 1520    [ 'Loaded ~D graphs (~D triples) in ~2f sec. (~d% CPU = ~2f sec.)'-
 1521      [Graphs, Triples, Wall, Percent, Time] ].
 1522% attach_graph/2
 1523message(restore(true, Action)) -->
 1524    !,
 1525    silent_message(Action).
 1526message(restore(brief, Action)) -->
 1527    !,
 1528    brief_message(Action).
 1529message(restore(_, Graph)) -->
 1530    [ 'Restoring ~p ... '-[Graph], flush ].
 1531message(restore(_, snapshot(_))) -->
 1532    [ at_same_line, '(snapshot) '-[], flush ].
 1533message(restore(_, journal(_))) -->
 1534    [ at_same_line, '(journal) '-[], flush ].
 1535message(restore(_, done(_, Time, Count))) -->
 1536    [ at_same_line, '~D triples in ~2f sec.'-[Count, Time] ].
 1537% load_source/4
 1538message(restore(_, snapshot(G, _))) -->
 1539    [ 'Restoring ~p\t(snapshot)'-[G], flush ].
 1540message(restore(_, journal(G, _))) -->
 1541    [ 'Restoring ~p\t(journal)'-[G], flush ].
 1542message(restore(_, done(_, Time, Count))) -->
 1543    [ at_same_line, '~D triples in ~2f sec.'-[Count, Time] ].
 1544% journal handling
 1545message(update_failed(S,P,O,Action)) -->
 1546    [ 'Failed to update <~p ~p ~p> with ~p'-[S,P,O,Action] ].
 1547% directory reindexing
 1548message(reindex(Count, Depth)) -->
 1549    [ 'Restructuring database with ~d levels (~D graphs)'-[Depth, Count] ].
 1550message(reindex(Depth)) -->
 1551    [ 'Fixing database directory structure (~d levels)'-[Depth] ].
 1552message(read_only) -->
 1553    [ 'Cannot write persistent store; continuing in read-only mode.', nl,
 1554      'All changes to the RDF store will be lost if this process terminates.'
 1555    ].
 1556
 1557silent_message(_Action) --> [].
 1558
 1559brief_message(done(Graph, _Time, _Count, Nth, Total)) -->
 1560    { file_base_name(Graph, Base) },
 1561    [ at_same_line,
 1562      '\r~p~`.t ~D of ~D graphs~72|'-[Base, Nth, Total],
 1563      flush
 1564    ].
 1565brief_message(_) --> [].
 1566
 1567
 1568prolog:message_context(rdf_locked(Args)) -->
 1569    { memberchk(time(Time), Args),
 1570      memberchk(pid(Pid), Args),
 1571      format_time(string(S), '%+', Time)
 1572    },
 1573    [ nl,
 1574      'locked at ~s by process id ~w'-[S,Pid]
 1575    ]